type pointer = {evaluator: string; code: Concepts.Hash.hash}

type t =
  { name: string option;
    schematics:
      [ `Persisted of Concepts.Hash.hash
      | `Temporary of Protocols.Schematics.attribute_description BatMap.String.t ];
    pointer: pointer option }

module Error = struct
  open Concepts.Condition

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

  let unbound_program name =
    condition "unbound-program"
      "A relation generated by a program has to be given a state to run within before it is read"
      ("relation" |=| Concepts.Value.String name)

  let undecidable_membership () =
    condition "undecidable-membership"
      "A relation with no membership test and no end was asked whether it holds a tuple" empty
end

module Persisted = struct
  type nonrec t = {name: string; schematics: Concepts.Hash.hash; pointer: pointer}

  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 = 'E'
    let malformed = Error.malformed_relation

    let fields {name; schematics; pointer= {evaluator; code}} =
      let open Concepts.Encoding in
      [ "name", Field.string name;
        "schema", Value.bencode_of_hash schematics;
        "evaluator", Field.string evaluator;
        "program", Value.bencode_of_hash code ]

    let of_fields fields =
      let open Utilities.Result in
      let open Concepts.Encoding in
      let* name = Field.require_string "name" fields in
      let* schematics = Bencode.field "schema" fields |> fmap Value.hash_of_bencode in
      let* evaluator = Field.require_string "evaluator" fields in
      let* code = Bencode.field "program" fields |> fmap Value.hash_of_bencode in
      Ok {name; schematics; pointer= {evaluator; code}}
  end
end

(* FIXME: Bring the Persisted/Temporary out to avoid using the
   wildcard and optionals *)
let persisted = function
  | {name= Some name; schematics= `Persisted schematics; pointer= Some pointer} ->
      Some {Persisted.name; schematics; pointer}
  | _ -> None

class ephemeral_relation ?program ?decide description value modes enumerate inputs =
  object (self)
    inherit Lifecycle.counted
    inherit Identity.of_id
    method to_string = "ephemeral-relation"
    val relation : t = value
    method destroy = List.iter Protocols.Handle.release inputs
    method enumerate : (Protocols.Handle.t, Concepts.Condition.condition) result = enumerate ()
    method modes : (Concepts.Mode.t, Concepts.Condition.condition) result = modes ()
    method describe () = Ok (Protocols.Schematics.Relation description)

    method contains (tuple : Concepts.Tuple.t) =
      let open Utilities.Result in
      match decide with
      | Some decide -> decide tuple
      | None ->
          let* modes = self#modes in
          if not (Concepts.Mode.exhaustible modes BatSet.String.empty) then
            Error (Error.undecidable_membership ())
          else
            let* cursor = self#enumerate in
            Protocols.Handle.releasing cursor (fun () ->
                let* scan = Protocols.Cursor.require cursor in
                Protocols.Cursor.exists scan (fun member ->
                    Concepts.Hash.hash_equals (Concepts.Tuple.hash member)
                      (Concepts.Tuple.hash tuple) ) )

    method protocols : Protocols.Handle.protocol list =
      let protocols =
        Protocols.[Relation.make self; Enumerable.make self; Schematics.make self]
        @ Option.fold ~none:[] ~some:Protocols.Handle.protocols program
      in
      match persisted relation with
      | None -> protocols
      | Some record ->
          Protocols.Addressable.make
            object
              method address = Persisted.Representation.to_blob record |> Concepts.Hash.hash_of_blob
            end
          :: protocols
  end

let instantiate ?(inputs = []) ?program ?decide ~modes description value enumerate =
  new ephemeral_relation ?program ?decide description value modes enumerate inputs
  |> Protocols.Handle.make

let literal description tuples =
  instantiate description
    ~modes:(fun () -> Ok [Concepts.(Mode.mode [] (Cardinality.bounded (List.length tuples)))])
    {name= None; schematics= `Temporary description; pointer= None}
    (fun () -> Ok (Generator.cursor_of (fun ~yield -> List.iter yield tuples; Ok ())))

type binding = {root: Protocols.Handle.t; within: Protocols.Handle.t}

(* The relation a program produced has to outlive the cursor over it. *)
let enumerate_owned relation =
  let open Utilities.Result in
  let owner =
    object
      method reference = true
      method release = Protocols.Handle.release relation
      method is_managed = true
    end
  in
  match
    let* enumerable = Protocols.Enumerable.require relation in
    Protocols.Enumerable.enumerate enumerable
  with
  | Error c -> Protocols.Handle.release relation; Error c
  | Ok cursor ->
      Ok
        (Protocols.Handle.make
           object
             inherit Lifecycle.retaining cursor owner
             method to_string = Protocols.Handle.to_string cursor
           end )

let asking f relation =
  Protocols.Handle.releasing relation (fun () -> Result.bind (Protocols.Relation.require relation) f)

let generated ?program ?within name description value =
  let invoking f () =
    let open Utilities.Result in
    match program, within with
    | Some program, Some within ->
        let* executable = Protocols.Executable.require program in
        Result.bind (Protocols.Executable.invoke executable [] ~within) f
    | _ -> Error (Error.unbound_program name)
  in
  instantiate ?program
    ~decide:(fun tuple -> invoking (asking (fun r -> Protocols.Relation.contains r tuple)) ())
    ~modes:(invoking (asking Protocols.Relation.modes))
    description value (invoking enumerate_owned)

let of_program ?within description program =
  generated ~program ?within (Protocols.Handle.to_string program) description
    {name= None; schematics= `Temporary description; pointer= None}

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

  let restore ?program ?within ({Persisted.name; schematics; pointer} : Persisted.t)
      description =
    generated ?program ?within name description
      {name= Some name; schematics= `Persisted schematics; pointer= Some pointer}

  let load ?bind connection (record : Persisted.t) =
    let open Utilities.Result in
    let* description, source =
      SI.with_read connection (fun tx ->
          let* description = R.schema_of tx record.schematics in
          let* code = SI.get_req tx (S.Hash record.pointer.code) in
          Ok (description, Bytes.to_string (Concepts.Blob.bytes_of_blob code)) )
    in
    match bind with
    | None -> Ok (restore record description)
    | Some {root; within} ->
        let* program = Program.load root ~evaluator:record.pointer.evaluator ~source in
        Ok (restore ~program ~within record description)

  let store tx ~name ~evaluator ~source description =
    let open Utilities.Result in
    let* schematics = R.store_schema tx description in
    let* address = SI.store_blob tx (Concepts.Blob.blob_of_bytes (Bytes.of_string source)) in
    let record = {Persisted.name; schematics; pointer= {evaluator; code= address}} in
    let* _ = SI.store_blob tx (Persisted.Representation.to_blob record) in
    Ok record

  let persist connection ~name program description =
    let open Utilities.Result in
    let* {Program.evaluator; source; _} = Program.require program in
    SI.with_transaction connection (fun tx ->
        let* record = store tx ~name ~evaluator ~source description in
        Ok (restore ~program record description) )
end