path: root/lib/listener.ml

module Make =
functor
  (T : Transport.TRANSPORT)
  (NT : Nt.S)
  ->
  struct
    module type SubS = Sublanguage.S

    module Sess = Session.Make (NT)
    module B    = Branch.Make (NT)

    module Error = struct
      open Condition

      let syntax_error msg =
        condition "syntax-error" msg empty

      let unrecognized_sublanguage tag =
        condition "unrecognized-sublanguage" "Unrecognized sublanguage"
          ("tag" |=| (of_string tag))

      let malformed_expression expr =
        condition "malformed-expression" "Malformed expression"
          ("expression" |=| (of_sexp expr))
    end

    (** Per-connection state: session owns the active branch. The
        branch's [relation_path] becomes the context resolver, so all
        sublanguages address the full RNT namespace rather than a
        single multigroup. **)
    type conn_state = {
      claims  : Nt.claims;
      session : Sess.session;
    }

    (** Build a fresh execution context from the current session
        state. Called at the top of every handle_sublanguage so it
        always reflects the live branch after any prior VCL
        switch. **)
    let make_ctx (state : conn_state) : Sublanguage_context.t =
      let br      = state.session#branch in
      let claims  = state.claims in
      let session = state.session in
      {
        Sublanguage_context.write_handle  = br#branch_handle;
        branch                            = (br :> Sublanguage_context.branch_view);
        resolve                           = br#path;
        switch_branch = fun name ->
          match br#close () with
          | Error e -> Error e
          | Ok () ->
              begin match B.open_branch claims name with
               | Error e ->
                  begin match B.open_branch claims "master" with
                  | Ok mbr -> session#set_branch mbr
                  | Error _ -> ()
                  end;
                  Error e
               | Ok new_br ->
                   session#set_branch new_br;
                   Ok new_br#multigroups
              end
      }

    let read_command input =
      try Ok (Sexplib.Sexp.input_sexp input)
      with Sexplib.Sexp.Parse_error { err_msg; _ } ->
        Error (Error.syntax_error err_msg)

    let sublanguages =
      List.fold_right
        (fun (module Language : SubS) ->
          Utilities.StringMap.add Language.name (module Language : SubS))
        [
          (module Drl.Sublanguage.Make (NT) : SubS);
          (module Ddl.Sublanguage.Make (NT) : SubS);
          (module Dml.Sublanguage.Make (NT) : SubS);
          (module Icl.Sublanguage.Make (NT) : SubS);
          (module Prl.Sublanguage.Make (NT) : SubS);
          (module Scl.Sublanguage.Make (NT) : SubS);
          (module Vcl.Sublanguage.Make (NT) : SubS);
        ]
        Utilities.StringMap.empty

    let find_language tag =
      Utilities.StringMap.find_opt tag sublanguages
      |> Option.to_result ~none:(Error.unrecognized_sublanguage tag)

    let execute_sublanguage ctx expr (module Language : SubS) =
      Language.parse_sexp expr
      |> Utilities.Result.fmap (Language.execute ctx)

    let execute_command ctx = function
      | Sexplib.Sexp.(List [ Atom tag; expr ]) ->
          find_language tag |> Utilities.Result.fmap (execute_sublanguage ctx expr)
      | s -> Error (Error.malformed_expression s)

    let tuple_to_sexp (t : Tuple.materialized) =
      let open Sexplib.Sexp in
      List
        (Tuple.AttributeMap.bindings t.Tuple.attributes
        |> List.map (fun (k, attr) ->
            List [ Atom k; Conventions.AbstractValue.sexp_of_t attr.Attribute.value ]))

    let output_response out_ch sexp =
      let response = Sexplib.Sexp.to_string sexp in
      output_string out_ch response;
      output_string out_ch "\n";
      flush out_ch;
      Utilities.print_with_time response

    let send_error out_ch e =
      output_response out_ch
        Sexplib.Sexp.(List [ Atom "error"; Condition.to_sexp e ])

    (* Serializes a sublanguage result.  [branch] is the write target whose
       tip and name surface in Cursor / Transition responses. *)
    let serialize (branch : B.branch) =
      let open Sexplib.Sexp in
      function
      | Error e -> List [ Atom "error"; Condition.to_sexp e ]
      | Ok (Sublanguage_types.Cursor { cursor_id; rows; has_more }) ->
          let row_sexps = List.map tuple_to_sexp rows in
          List
            [
              Atom "cursor";
              List [ Atom "id";        Atom cursor_id ];
              List [ Atom "rows";      List row_sexps ];
              List [ Atom "row_count"; Atom (string_of_int (List.length row_sexps)) ];
              List [ Atom "has_more";  Atom (string_of_bool has_more) ];
              List [ Atom "snapshot";  Atom branch#tip ];
              List [ Atom "branch";    Atom branch#name ];
            ]
      | Ok (Sublanguage_types.Query _rel) ->
          List [ Atom "error"; Atom "unexpected Query result: use Cursor path" ]
      | Ok (Sublanguage_types.Transition _new_cache) ->
          List
            [
              Atom "ok";
              List [ Atom "snapshot"; Atom branch#tip ];
              List [ Atom "branch";   Atom branch#name ];
            ]
      | Ok (Sublanguage_types.SessionSwitch name) ->
          List [ Atom "ok"; List [ Atom "message"; Atom ("Switched to branch " ^ name) ] ]
      | Ok (Sublanguage_types.CreateMultigroup name) ->
          List [ Atom "ok"; List [ Atom "message"; Atom ("Multigroup " ^ name ^ " created") ] ]

    let handle_sublanguage output state sexp =
      let ctx = make_ctx !state in
      match execute_command ctx sexp with
      | Error e -> send_error output e
      | Ok result ->
          (* Sync updated schema_cache back into the branch mirror on any
             state-advancing result (DDL, ICL, VCL).  For VCL the session
             already holds the new branch; refresh_mg is a no-op on it but
             correct for DDL/ICL which stay on the same branch. *)
          (match result with
           | Sublanguage_types.Transition delta ->
               !state.session#branch#apply_delta delta
           | _ -> ());
          serialize !state.session#branch (Ok result) |> output_response output

    let handle_client connection =
      let input  = T.input connection in
      let output = T.output connection in
      match NT.authenticate Nt.PlainText with
      | Error e ->
          (* FIXME: why is this a syntax error? *)
          send_error output e
      | Ok claims ->
          (match Sess.open_session claims ~branch_name:"master" with
           | Error e ->
               send_error output e
           | Ok sess ->
               let state = ref { claims; session = sess } in
               (try
                 while true do
                   match read_command input with
                   | Error e -> send_error output e
                   | Ok sexp -> handle_sublanguage output state sexp
                 done
               with
               | End_of_file -> ignore (!state.session#close ())
               | e ->
                   ignore (!state.session#close ());
                   Printf.eprintf "Error handling connection: %s" (Printexc.to_string e)))

    let spawn_handler connection =
      Stdlib.Domain.spawn (fun () -> handle_client connection)

    let run transport =
      T.listen transport;
      while true do
        T.accept transport |> spawn_handler |> ignore
      done
  end