open Relational_engine let default_port = 7777 let default_limit = 50 (* 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) (* TODO: registration failures are silently ignored; a broken prelude relation leaves the catalog in a partially-seeded state with no indication to the user. *) 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 ~schema:rel.schema ~generator:(Option.get rel.generator) ~membership_criteria:rel.membership_criteria ~cardinality:rel.cardinality with | Ok (new_db, _) -> new_db | Error e -> Printf.eprintf "Warning: failed to register prelude relation %s: %s\n%!" rel.name (Manipulation.string_of_error e); db) db [ less_than_natural ; less_than_or_equal_natural ; greater_than_natural ; greater_than_or_equal_natural ; equal_natural ; not_equal_natural ; plus_natural ; times_natural ; minus_natural ; 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 setup () = let ( let* ) = Result.bind in let* storage = Management.Physical.Memory.create () |> Result.map_error (Fun.const "Failed to create storage") in let* db = Manipulation.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 () = match setup () with | Error msg -> failwith 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