Add src/hk_api_events_ws_h.erl

src/hk_api_events_ws_h.erl

@@ -0,0 +1,64 @@
+%%%-------------------------------------------------------------------
+%%% @doc `/events' API domain, real-time half: `GET /api/events/stream'
+%%% upgrades to a WebSocket and pushes every `hk_event' published
+%%% system-wide, in publish order, for as long as the connection is
+%%% open. This is what "support streaming events to the frontend
+%%% using an appropriate real-time transport" means concretely --
+%%% WebSocket over the same Cowboy listener as the REST API, backed
+%%% directly by `hk_event_bus:subscribe/1'.
+%%%
+%%% An optional `?category_prefix=agent.' or `?subject_type=agent&
+%%% subject_id=...' query string scopes the subscription (see
+%%% `hk_event_bus:matches/2') -- used by e.g. the agent-console view
+%%% to watch one agent without receiving the whole system's traffic.
+%%% @end
+%%%-------------------------------------------------------------------
+-module(hk_api_events_ws_h).
+-behaviour(cowboy_websocket).
+
+-export([init/2, websocket_init/1, websocket_handle/2, websocket_info/2, terminate/3]).
+
+init(Req, _State) ->
+    Qs = cowboy_req:parse_qs(Req),
+    Filter = build_filter(Qs),
+    {cowboy_websocket, Req, #{filter => Filter}}.
+
+websocket_init(State = #{filter := Filter}) ->
+    ok = hk_event_bus:subscribe(Filter),
+    ok = hk_frontend_client_registry:client_connected(self()),
+    {[{text, jsx:encode(#{type => <<"connected">>})}], State}.
+
+websocket_handle({text, <<"ping">>}, State) ->
+    {[{text, <<"pong">>}], State};
+websocket_handle(_Frame, State) ->
+    {[], State}.
+
+websocket_info({hk_event, EventMap}, State) ->
+    {[{text, jsx:encode(#{type => <<"event">>, event => EventMap})}], State};
+websocket_info(_Info, State) ->
+    {[], State}.
+
+terminate(_Reason, _Req, _State) ->
+    catch hk_event_bus:unsubscribe(),
+    catch hk_frontend_client_registry:client_disconnected(self()),
+    ok.
+
+%%% internal
+
+build_filter(Qs) ->
+    Raw = lists:foldl(
+        fun ({<<"category_prefix">>, V}, Acc) -> Acc#{category_prefix => V};
+            ({<<"subject_type">>, V}, Acc) -> Acc#{subject_type => binary_to_atom(V, utf8)};
+            ({<<"subject_id">>, V}, Acc) -> Acc#{subject_id => V};
+            (_, Acc) -> Acc
+        end, #{}, Qs),
+    finalize_subject_filter(Raw).
+
+finalize_subject_filter(Filter) ->
+    case {maps:take(subject_type, Filter), maps:is_key(subject_id, Filter)} of
+        {{Type, Filter2}, true} ->
+            {SubjectId, Filter3} = maps:take(subject_id, Filter2),
+            Filter3#{subject => {Type, SubjectId}};
+        _ ->
+            Filter
+    end.