New Server & Manipulation refactor

Co-authored-by: Estevan Castilho <tevo@riajuu.net>

2 files changed, 44 insertions, 260 deletions

diff --git a/bin/main.ml b/bin/main.ml
deleted file mode 100644
index 6b2a068..0000000
--- a/bin/main.ml
+++ /dev/null
@@ -1,17 +0,0 @@
-let main () =
-  let digest =
-    Sha256.to_hex (Sha256.string "sakura")
-  in
-  print_endline digest
-
-let () =
-  let tuple =
-    Relational_engine.Tuple.make_materialized ~relation:"natural_less_than"
-      ~attributes:
-        (Relational_engine.Tuple.AttributeMap.of_list
-           [ ("left", Obj.magic 0); ("right", Obj.magic 1) ])
-  in
-  print_endline @@ string_of_bool
-  @@ Relational_engine.Prelude.Standard.less_than_natural.membership_criteria
-       tuple;
-  main ()
diff --git a/bin/server.ml b/bin/server.ml
index 82965dc..1d4d5b3 100644
--- a/bin/server.ml
+++ b/bin/server.ml
@@ -1,160 +1,21 @@
 open Relational_engine
 
-let default_port = 7777
-let default_limit = 50
+module B = Management.Physical.Memory
+module T = Transport.TCP
+module Listener = Listener.Make(T)(B)
 
-(* TODO: AbstractValue carries no type tag of its own; we recover the OCaml
-   runtime tag here via Obj introspection. This is fragile — it breaks the
-   moment a domain stores a boxed integer or a custom block. The right fix is
-   a typed value representation in Conventions. *)
-
-(* TODO: Generator.Error is silently treated as end-of-stream here, hiding
-   real failures from the client. *)
-let materialize_generator gen limit =
-  let rec go gen pos acc count =
-    if count >= limit then (List.rev acc, true)
-    else
-      match gen (Some pos) with
-      | Generator.Done -> (List.rev acc, false)
-      | Generator.Error _ -> (List.rev acc, false)
-      | Generator.Value (t, next) -> (
-          let mat =
-            match t with
-            | Tuple.Materialized m -> Some m
-            | Tuple.NonMaterialized _ -> None
-          in
-          match mat with
-          | None -> go next (pos + 1) acc count
-          | Some m -> go next (pos + 1) (m :: acc) (count + 1))
-  in
-  go gen 0 [] 0
-
-let materialize_relation storage (rel : Relation.t) limit =
-  let gen = Algebra.Memory.to_generator storage rel in
-  materialize_generator gen limit
-
-let short_hash (h : string) = String.sub h 0 (min 8 (String.length h))
-
-module BranchOps = Management.Branch.Make (Management.Physical.Memory)
-module Dispatch = Dbms_language.Memory
-
-let get_branch storage =
-  match BranchOps.get_head storage with Ok (Some name) -> name | _ -> "--"
-
-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 relation_to_sexp storage db_name db_hash (rel : Relation.t) limit =
-  let open Sexplib.Sexp in
-  let schema_sexp =
-    List (List.map (fun (a, d) -> List [ Atom a; Atom d ]) rel.Relation.schema)
-  in
-  let tuples, truncated = materialize_relation storage rel limit in
-  let rows_sexp = List (List.map tuple_to_sexp tuples) in
-  to_string
-  @@ List
-       [
-         Atom "relation";
-         List [ Atom "name"; Atom rel.Relation.name ];
-         List [ Atom "schema"; schema_sexp ];
-         List [ Atom "rows"; rows_sexp ];
-         List [ Atom "row_count"; Atom (string_of_int (List.length tuples)) ];
-         List [ Atom "truncated"; Atom (string_of_bool truncated) ];
-         List [ Atom "db_hash"; Atom (short_hash db_hash) ];
-         List [ Atom "db_name"; Atom db_name ];
-         List [ Atom "branch"; Atom (get_branch storage) ];
-       ]
-
-let ok_response storage db_name db_hash msg =
-  let open Sexplib.Sexp in
-  to_string
-  @@ List
-       [
-         Atom "ok";
-         List [ Atom "message"; Atom msg ];
-         List [ Atom "db_hash"; Atom (short_hash db_hash) ];
-         List [ Atom "db_name"; Atom db_name ];
-         List [ Atom "branch"; Atom (get_branch storage) ];
-       ]
-
-let error_response storage db_name db_hash err_sexp =
-  let open Sexplib.Sexp in
-  to_string
-  @@ List
-       [
-         Atom "error";
-         err_sexp;
-         List [ Atom "db_hash"; Atom (short_hash db_hash) ];
-         List [ Atom "db_name"; Atom db_name ];
-         List [ Atom "branch"; Atom (get_branch storage) ];
-       ]
-
-let advance_head_branch storage new_hash =
-  match BranchOps.get_head storage with
-  | Ok (Some branch_name) ->
-      ignore (BranchOps.update_tip storage ~name:branch_name ~tip:new_hash)
-  | _ -> ()
-
-let dbms_dispatch =
-  Dispatch.create
-    [
-      (module Drl.Sublanguage.Memory : Dispatch.SubS);
-      (module Ddl.Sublanguage.Memory : Dispatch.SubS);
-      (module Dml.Sublanguage.Memory : Dispatch.SubS);
-      (module Icl.Sublanguage.Memory : Dispatch.SubS);
-      (module Dcl.Sublanguage.Memory : Dispatch.SubS);
-      (module Scl.Sublanguage.Memory : Dispatch.SubS);
-    ]
-
-let cursor_to_sexp storage db_name db_hash cursor_id rows has_more =
-  let open Sexplib.Sexp in
-  let row_sexps = List.map tuple_to_sexp rows in
-  let rows_sexp = List row_sexps in
-  to_string
-  @@ List
-       [
-         Atom "cursor";
-         List [ Atom "id"; Atom cursor_id ];
-         List [ Atom "rows"; rows_sexp ];
-         List [ Atom "row_count"; Atom (string_of_int (List.length row_sexps)) ];
-         List [ Atom "has_more"; Atom (string_of_bool has_more) ];
-         List [ Atom "db_hash"; Atom (short_hash db_hash) ];
-         List [ Atom "db_name"; Atom db_name ];
-         List [ Atom "branch"; Atom (get_branch storage) ];
-       ]
-
-let execute_command storage db_ref cmd =
-  let db = !db_ref in
-  let h = db.Management.Database.hash in
-  let name = db.Management.Database.name in
-  let ok = ok_response storage name in
-  let err = error_response storage name in
-  match Dispatch.execute dbms_dispatch storage db cmd with
-  | Ok (Sublanguage.Query rel) ->
-      relation_to_sexp storage name h rel default_limit
-  | Ok (Sublanguage.Transition (new_db, msg)) ->
-      db_ref := new_db;
-      advance_head_branch storage new_db.Management.Database.hash;
-      ok new_db.Management.Database.hash msg
-  | Ok (Sublanguage.Cursor { cursor_id; rows; has_more }) ->
-      cursor_to_sexp storage name h cursor_id rows has_more
-  | Error e -> err h (Dispatch.sexp_of_dispatch_error e)
+module Memory = Manipulation.Make (B)
 
 (* TODO: registration failures are silently ignored; a broken prelude
    relation leaves the catalog in a partially-seeded state with no
    indication to the user. *)
+(* TODO: think of a better way to initialize new databases *)
 let register_prelude_relations storage db =
   let open Prelude.Standard in
   List.fold_left
     (fun db (rel : Relation.t) ->
       match
-        Manipulation.Memory.create_immutable_relation storage db ~name:rel.name
+        Memory.create_immutable_relation storage db ~name:rel.name
           ~schema:rel.schema ~generator:(Option.get rel.generator)
           ~membership_criteria:rel.membership_criteria
           ~cardinality:rel.cardinality
@@ -163,7 +24,7 @@ let register_prelude_relations storage db =
       | Error e ->
           Printf.eprintf
             "Warning: failed to register prelude relation %s: %s\n%!" rel.name
-            (Manipulation.string_of_error e);
+            (Sexplib.Sexp.to_string (Error.sexp_of_error e));
           db)
     db
     [
@@ -179,43 +40,27 @@ let register_prelude_relations storage db =
       divide_natural;
     ]
 
-let print_with_time str =
-  let now = Unix.gettimeofday () in
-  let tm = Unix.localtime now in
-  let orange = "\027[38;5;208m" in
-  let reset = "\027[0m" in
-  let formatted_time =
-    Printf.sprintf "%04d-%02d-%02d %02d:%02d:%02d" (tm.Unix.tm_year + 1900)
-      (tm.Unix.tm_mon + 1) tm.Unix.tm_mday tm.Unix.tm_hour tm.Unix.tm_min
-      tm.Unix.tm_sec
-  in
-  print_endline @@ Printf.sprintf "%s[%s]%s %s" orange formatted_time reset str
-
-(* TODO: single-threaded accept loop — one slow or hung client blocks all
-   others. Should use threads or non-blocking I/O. *)
-let handle_client storage db_ref fd =
-  let ic = Unix.in_channel_of_descr fd in
-  let oc = Unix.out_channel_of_descr fd in
-  try
-    while true do
-      let line = input_line ic in
-      let line = String.trim line in
-      if line <> "" then begin
-        let response =
-          try execute_command storage db_ref line
-          with e ->
-            let db = !db_ref in
-            error_response storage db.Management.Database.name
-              db.Management.Database.hash
-              Sexplib.Sexp.(
-                List [ Atom "uncaught-error"; Atom (Printexc.to_string e) ])
-        in
-        print_with_time response;
-        output_string oc (response ^ "\n");
-        flush oc
-      end
-    done
-  with End_of_file | Unix.Unix_error _ -> ()
+(* let register_branch_relations storage db = *)
+(*   (\* Register ephemeral sakura:branch and sakura:head relations backed by *)
+(*      the raw branch registry — no stored tuples, no circularity. *\) *)
+(*   let register db rel = *)
+(*     match *)
+(*       Memory.create_immutable_relation storage db *)
+(*         ~name:rel.Relation.name ~schema:rel.Relation.schema *)
+(*         ~generator:(Option.get rel.Relation.generator) *)
+(*         ~membership_criteria:rel.Relation.membership_criteria *)
+(*         ~cardinality:rel.Relation.cardinality *)
+(*     with *)
+(*     | Ok (db, _) -> db *)
+(*     | Error e -> *)
+(*         Printf.eprintf "Warning: failed to register %s: %s\n%!" *)
+(*           rel.Relation.name *)
+(*           (Manipulation.Error.string_of_error e); *)
+(*         db *)
+(*   in *)
+(*   db *)
+(*   |> Fun.flip register (BranchOps.branch_relation storage) *)
+(*   |> Fun.flip register (BranchOps.head_relation storage) *)
 
 let setup () =
   let ( let* ) = Result.bind in
@@ -224,71 +69,27 @@ let setup () =
     |> Result.map_error (Fun.const "Failed to create storage")
   in
   let* db =
-    Manipulation.Memory.create_database storage ~name:"sakura"
+    Memory.create_database storage ~name:"sakura"
     |> Result.map_error (Fun.const "Failed to create initial database")
   in
   Ok (storage, db)
 
-let register_branch_relations storage db =
-  (* Register ephemeral sakura:branch and sakura:head relations backed by
-     the raw branch registry — no stored tuples, no circularity. *)
-  let register db rel =
-    match
-      Manipulation.Memory.create_immutable_relation storage db
-        ~name:rel.Relation.name ~schema:rel.Relation.schema
-        ~generator:(Option.get rel.Relation.generator)
-        ~membership_criteria:rel.Relation.membership_criteria
-        ~cardinality:rel.Relation.cardinality
-    with
-    | Ok (db, _) -> db
-    | Error e ->
-        Printf.eprintf "Warning: failed to register %s: %s\n%!"
-          rel.Relation.name
-          (Manipulation.string_of_error e);
-        db
-  in
-  db
-  |> Fun.flip register (BranchOps.branch_relation storage)
-  |> Fun.flip register (BranchOps.head_relation storage)
-
-let warn_on_error fmt = function Ok () -> () | Error e -> Printf.eprintf fmt e
-let cursor_gc_max_age = 300.0 (* 5 minutes *)
-let cursor_gc_interval = 60.0 (* run GC at most once per minute *)
+let address = "127.0.0.1"
+let port = 7777
 
 let () =
   match setup () with
-  | Error msg -> failwith msg
+  | Error msg -> Printf.eprintf "Couldn't initialize: %s" msg
   | Ok (storage, db) ->
-      let db =
-        db
-        |> register_prelude_relations storage
-        |> register_branch_relations storage
-      in
-      (* Initialise cursor session registry for SCL *)
-      let session_reg = Session.create () in
-      Scl.Executor.set_sessions session_reg;
-      (* Create default master branch in raw registry and set HEAD *)
-      BranchOps.create storage ~name:"master" ~tip:db.Management.Database.hash
-      |> warn_on_error "Warning: failed to create master branch: %s\n%!";
-      BranchOps.checkout storage "master"
-      |> warn_on_error "Warning: failed to checkout master: %s\n%!";
-      let db_ref = ref db in
-      let last_gc = ref 0.0 in
-      let sock = Unix.socket Unix.PF_INET Unix.SOCK_STREAM 0 in
-      Unix.setsockopt sock Unix.SO_REUSEADDR true;
-      Unix.setsockopt sock Unix.SO_REUSEPORT true;
-      Unix.bind sock (Unix.ADDR_INET (Unix.inet_addr_loopback, default_port));
-      Unix.listen sock 5;
-      Printf.printf "Sakura server listening on 127.0.0.1:%d\n" default_port;
-      Printf.printf "Database hash: %s\n%!"
-        (short_hash db.Management.Database.hash);
-      while true do
-        let client_fd, _addr = Unix.accept sock in
-        handle_client storage db_ref client_fd;
-        (try Unix.close client_fd with _ -> ());
-        let now = Unix.gettimeofday () in
-        if now -. !last_gc >= cursor_gc_interval then begin
-          Session.gc session_reg ~max_age:cursor_gc_max_age;
-          last_gc := now
-        end
-      done
+     let db =
+       db
+       |> register_prelude_relations storage
+       (* |> register_branch_relations storage *)
+     in
+     let transport =
+       T.create
+         { address_family = Unix.PF_INET;
+           address = Unix.ADDR_INET (Unix.inet_addr_of_string address, port) }
+     in
+     Printf.printf "Listening on %s:%d" address port; print_newline ();
+     Listener.run transport db storage