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 SC = Sublanguage_context.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 ({session; claims} : conn_state) : Sublanguage_context.t =
      SC.make_ctx session claims

    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