path: root/bin/server.ml

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