%%% @doc XML Server for Lazy Relation Consumption %%% %%% Provides a TCP server that accepts query plans as Erlang term strings %%% and returns relation tuples as XML. The server maintains lazy iterators %%% and allows clients to consume tuples on demand. %%% %%% == Protocol == %%% %%% Client sends commands as line-delimited strings: %%% %%% ``` %%% QUERY - Execute a query plan, returns session ID %%% NEXT - Get next N tuples from iterator %%% CLOSE - Close iterator and free resources %%% ``` %%% %%% Server responds with XML: %%% %%% ``` %%% %%% %%% abc123 %%% %%% %%% 1 %%% Alice %%% %%% %%% %%% ``` %%% %%% == Usage == %%% %%% ``` %%% % Start server on port 8080 %%% xml_server:start(8080). %%% %%% % Client connects and sends: %%% % QUERY {scan, employees}\n %%% % NEXT 10\n %%% % NEXT 10\n %%% % CLOSE \n %%% ``` %%% %%% @author Nekoma Team %%% @copyright 2025 -module(xml_server). -export([start/1, start/2, stop/0]). -export([init/2, handle_client/2]). -include("operations.hrl"). %%% Public API %% @doc Start the XML server on specified port with example database. %% %% @param Port TCP port number to listen on %% @returns {ok, Pid} on success -spec start(pos_integer()) -> {ok, pid()}. start(Port) -> DB = repl:example_db(), start(Port, DB). %% @doc Start the XML server on specified port with custom database. %% %% @param Port TCP port number to listen on %% @param DB Database state to use for queries %% @returns {ok, Pid} on success -spec start(pos_integer(), term()) -> {ok, pid()}. start(Port, DB) -> spawn(?MODULE, init, [Port, DB]), io:format("XML Server started on port ~p~n", [Port]), {ok, whereis(xml_server)}. %% @doc Stop the XML server. %% %% @returns ok -spec stop() -> ok. stop() -> case whereis(xml_server) of undefined -> ok; Pid -> Pid ! stop, ok end. %%% Internal Functions %% @private %% @doc Initialize server and start listening. init(Port, DB) -> register(xml_server, self()), {ok, ListenSocket} = gen_tcp:listen(Port, [ binary, {packet, line}, {active, false}, {reuseaddr, true} ]), io:format("Listening for connections...~n"), accept_loop(ListenSocket, DB). %% @private %% @doc Accept incoming client connections. accept_loop(ListenSocket, DB) -> receive stop -> gen_tcp:close(ListenSocket), io:format("Server stopped~n"), ok after 0 -> case gen_tcp:accept(ListenSocket, 1000) of {ok, ClientSocket} -> io:format("Client connected: ~p~n", [ClientSocket]), spawn(?MODULE, handle_client, [ClientSocket, DB]), accept_loop(ListenSocket, DB); {error, timeout} -> accept_loop(ListenSocket, DB); {error, Reason} -> io:format("Accept error: ~p~n", [Reason]), accept_loop(ListenSocket, DB) end end. %% @private %% @doc Handle client session with command processing. handle_client(Socket, DB) -> Sessions = #{}, client_loop(Socket, DB, Sessions). %% @private %% @doc Main client command processing loop. client_loop(Socket, DB, Sessions) -> case gen_tcp:recv(Socket, 0) of {ok, Data} -> Line = binary_to_list(Data), Command = string:trim(Line), io:format("Received command: ~s~n", [Command]), case parse_command(Command) of {query, QueryStr} -> {NewSessions, Response} = handle_query(QueryStr, DB, Sessions), send_response(Socket, Response), client_loop(Socket, DB, NewSessions); {next, SessionId, Count} -> {NewSessions, Response} = handle_next(SessionId, Count, Sessions), send_response(Socket, Response), client_loop(Socket, DB, NewSessions); {close, SessionId} -> {NewSessions, Response} = handle_close(SessionId, Sessions), send_response(Socket, Response), client_loop(Socket, DB, NewSessions); {schema} -> Response = handle_schema(DB), send_response(Socket, Response), client_loop(Socket, DB, Sessions); {error, Msg} -> Response = error_response(Msg), send_response(Socket, Response), client_loop(Socket, DB, Sessions) end; {error, closed} -> io:format("Client disconnected~n"), cleanup_sessions(Sessions), ok; {error, Reason} -> io:format("Receive error: ~p~n", [Reason]), cleanup_sessions(Sessions), ok end. %% @private %% @doc Parse client command. parse_command(Command) -> case string:split(Command, " ", leading) of ["QUERY", QueryStr] -> {query, QueryStr}; ["NEXT", Rest] -> case string:split(Rest, " ", leading) of [SessionId, CountStr] -> try Count = list_to_integer(CountStr), {next, SessionId, Count} catch _:_ -> {error, "Invalid count"} end; _ -> {error, "Invalid NEXT syntax"} end; ["CLOSE", SessionId] -> {close, SessionId}; ["SCHEMA"] -> {schema}; _ -> {error, "Unknown command"} end. %% @private %% @doc Handle QUERY command - execute query and create iterator. handle_query(QueryStr, DB, Sessions) -> try % Parse and evaluate the query plan from string using erl_eval % This allows parsing of function literals {ok, Tokens, _} = erl_scan:string(QueryStr ++ "."), {ok, Exprs} = erl_parse:parse_exprs(Tokens), {value, QueryPlan, _} = erl_eval:exprs(Exprs, erl_eval:new_bindings()), % Prepare and execute query to get iterator PreparedPlan = query_planner:prepare(QueryPlan), Iterator = query_planner:execute(DB, PreparedPlan), % Generate session ID SessionId = generate_session_id(), % Store iterator in sessions NewSessions = maps:put(SessionId, Iterator, Sessions), % Return session ID Response = session_response(SessionId), {NewSessions, Response} catch Error:Reason:Stack -> io:format("Query error: ~p:~p~n~p~n", [Error, Reason, Stack]), {Sessions, error_response(io_lib:format("Query failed: ~p", [Reason]))} end. %% @private %% @doc Handle NEXT command - fetch next N tuples from iterator. handle_next(SessionId, Count, Sessions) -> case maps:get(SessionId, Sessions, undefined) of undefined -> {Sessions, error_response("Invalid session ID")}; Iterator -> try % Collect next Count tuples {Tuples, NewIterator} = collect_n_tuples(Iterator, Count), % Update session with new iterator state NewSessions = maps:put(SessionId, NewIterator, Sessions), % Return tuples as XML Response = tuples_response(Tuples), {NewSessions, Response} catch Error:Reason:Stack -> io:format("Next error: ~p:~p~n~p~n", [Error, Reason, Stack]), {Sessions, error_response(io_lib:format("Next failed: ~p", [Reason]))} end end. %% @private %% @doc Handle CLOSE command - close iterator and free resources. handle_close(SessionId, Sessions) -> case maps:get(SessionId, Sessions, undefined) of undefined -> {Sessions, error_response("Invalid session ID")}; Iterator -> operations:close_iterator(Iterator), NewSessions = maps:remove(SessionId, Sessions), Response = ok_response("Session closed"), {NewSessions, Response} end. %% @private %% @doc Handle SCHEMA command - return database schema with all relations and their attributes. handle_schema(DB) -> try % Get all relation names RelationNames = operations:get_relations(DB), % For each relation, get its hash and then the relation record Relations = lists:map(fun(Name) -> case operations:get_relation_hash(DB, Name) of {ok, Hash} -> % Read the relation record from Mnesia case mnesia:dirty_read(relation, Hash) of [Relation] -> {Name, Relation#relation.schema, Relation#relation.cardinality, Relation#relation.constraints, Relation#relation.provenance}; [] -> {Name, #{}, unknown, #{}, undefined} end; {error, _} -> {Name, #{}, unknown, #{}, undefined} end end, RelationNames), schema_response(Relations) catch Error:Reason:Stack -> io:format("Schema error: ~p:~p~n~p~n", [Error, Reason, Stack]), error_response(io_lib:format("Schema failed: ~p", [Reason])) end. %% @private %% @doc Collect N tuples from iterator lazily. collect_n_tuples(Iterator, Count) -> collect_n_tuples(Iterator, Count, []). collect_n_tuples(Iterator, 0, Acc) -> {lists:reverse(Acc), Iterator}; collect_n_tuples(Iterator, Count, Acc) -> case operations:next_tuple(Iterator) of {ok, Tuple} -> collect_n_tuples(Iterator, Count - 1, [Tuple | Acc]); done -> {lists:reverse(Acc), Iterator} end. %% @private %% @doc Generate unique session ID. generate_session_id() -> Timestamp = erlang:system_time(microsecond), Random = rand:uniform(1000000), Hash = crypto:hash(md5, term_to_binary({Timestamp, Random})), binary_to_list(base64:encode(Hash)). %% @private %% @doc Cleanup all sessions on disconnect. cleanup_sessions(Sessions) -> maps:foreach( fun(_SessionId, Iterator) -> operations:close_iterator(Iterator) end, Sessions). %%% XML Response Formatters %% @private %% @doc Format session response. session_response(SessionId) -> [ "\n", "\n", " ok\n", " ", SessionId, "\n", "\n" ]. %% @private %% @doc Format schema response as XML. schema_response(Relations) -> RelationsXML = lists:map(fun({Name, Schema, Cardinality, Constraints, Provenance}) -> % Convert cardinality to string CardStr = case Cardinality of {finite, N} -> io_lib:format("finite(~p)", [N]); aleph_zero -> "infinite"; continuum -> "continuum"; _ -> "unknown" end, % Convert schema map to attribute elements % Wrap each attribute in a list to keep it as a single element when reversed Attributes = maps:fold(fun(AttrName, Type, Acc) -> [ [" \n"] | Acc ] end, [], Schema), % Convert constraints to XML ConstraintsXML = case maps:size(Constraints) of 0 -> []; _ -> [ " \n", maps:fold(fun(AttrName, Constraint, Acc) -> ConstraintStr = format_constraint(Constraint), [[" ", ConstraintStr, "\n"] | Acc] end, [], Constraints), " \n" ] end, % Convert provenance to XML ProvenanceXML = case Provenance of undefined -> []; _ -> [ " ", format_provenance(Provenance), "\n" ] end, [ " \n", lists:reverse(Attributes), ConstraintsXML, ProvenanceXML, " \n" ] end, Relations), [ "\n", "\n", " ok\n", " \n", RelationsXML, " \n", "\n" ]. %% @private %% @doc Format constraint as string. format_constraint({eq, Value}) -> lists:flatten(io_lib:format("= ~p", [Value])); format_constraint({neq, Value}) -> lists:flatten(io_lib:format("!= ~p", [Value])); format_constraint({lt, Value}) -> lists:flatten(io_lib:format("< ~p", [Value])); format_constraint({lte, Value}) -> lists:flatten(io_lib:format("<= ~p", [Value])); format_constraint({gt, Value}) -> lists:flatten(io_lib:format("> ~p", [Value])); format_constraint({gte, Value}) -> lists:flatten(io_lib:format(">= ~p", [Value])); format_constraint({in, List}) -> lists:flatten(io_lib:format("in ~p", [List])); format_constraint({range, Min, Max}) -> lists:flatten(io_lib:format("in [~p, ~p]", [Min, Max])); format_constraint({member_of, Relation}) -> lists:flatten(io_lib:format("in ~p", [Relation])); format_constraint(Other) -> lists:flatten(io_lib:format("~p", [Other])). %% @private %% @doc Format provenance as string. format_provenance(undefined) -> "undefined"; format_provenance({base, Relation}) -> lists:flatten(io_lib:format("base(~p)", [Relation])); format_provenance({join, P1, P2}) -> lists:flatten(io_lib:format("join(~s, ~s)", [format_provenance(P1), format_provenance(P2)])); format_provenance({select, P, _Constraints}) -> lists:flatten(io_lib:format("select(~s)", [format_provenance(P)])); format_provenance({project, P, Attrs}) -> lists:flatten(io_lib:format("project(~s, ~p)", [format_provenance(P), Attrs])); format_provenance({take, P, N}) -> lists:flatten(io_lib:format("take(~s, ~p)", [format_provenance(P), N])); format_provenance(Other) -> lists:flatten(io_lib:format("~p", [Other])). %% @private %% @doc Format tuples response. tuples_response(Tuples) -> [ "\n", "\n", " ok\n", " \n", lists:map(fun tuple_to_xml/1, Tuples), " \n", "\n" ]. %% @private %% @doc Format single tuple as XML. tuple_to_xml(Tuple) -> Attributes = maps:to_list(Tuple), [ " \n", lists:map(fun({Name, Value}) -> EscapedValue = escape_xml(format_value(Value)), io_lib:format(" ~s\n", [Name, EscapedValue]) end, Attributes), " \n" ]. %% @private %% @doc Format value for XML output. format_value(Value) when is_integer(Value) -> integer_to_list(Value); format_value(Value) when is_float(Value) -> float_to_list(Value); format_value(Value) when is_atom(Value) -> atom_to_list(Value); format_value(Value) when is_binary(Value) -> binary_to_list(Value); format_value(Value) when is_list(Value) -> Value; format_value(Value) -> io_lib:format("~p", [Value]). %% @private %% @doc Escape XML special characters. escape_xml(Str) -> lists:flatten( lists:map( fun($<) -> "<"; ($>) -> ">"; ($&) -> "&"; ($") -> """; ($') -> "'"; (C) -> C end, lists:flatten(Str))). %% @private %% @doc Format error response. error_response(Message) -> [ "\n", "\n", " error\n", " ", escape_xml(Message), "\n", "\n" ]. %% @private %% @doc Format OK response. ok_response(Message) -> [ "\n", "\n", " ok\n", " ", escape_xml(Message), "\n", "\n" ]. %% @private %% @doc Send response to client. send_response(Socket, Response) -> gen_tcp:send(Socket, Response).