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