Stage 10 final

src/handlers/ws_handler.erl

@@ -0,0 +1,85 @@
+-module(ws_handler).
+-behaviour(cowboy_websocket).
+
+-export([init/2]).
+-export([websocket_init/1]).
+-export([websocket_handle/2]).
+-export([websocket_info/2]).
+-export([terminate/3]).
+
+-record(state, {
+  user_id :: binary() | undefined,
+  subscriptions = [] :: [binary()]
+}).
+
+init(Req, _Opts) ->
+  % Аутентификация через query параметр token
+  Qs = cowboy_req:parse_qs(Req),
+  case proplists:get_value(<<"token">>, Qs) of
+    undefined ->
+      {ok, cowboy_req:reply(401, #{}, <<"Missing token">>, Req), undefined};
+    Token ->
+      case logic_auth:verify_jwt(Token) of
+        {ok, Claims} ->
+          UserId = maps:get(<<"user_id">>, Claims),
+          {cowboy_websocket, Req, #state{user_id = UserId}};
+        {error, _} ->
+          {ok, cowboy_req:reply(401, #{}, <<"Invalid token">>, Req), undefined}
+      end
+  end.
+
+websocket_init(State) ->
+  % Регистрируем процесс в pg для получения уведомлений
+  pg:join(eventhub_ws, self()),
+  {ok, State}.
+
+websocket_handle({text, Msg}, State) ->
+  io:format("WebSocket received: ~s~n", [Msg]),
+  try jsx:decode(Msg, [return_maps]) of
+    #{<<"action">> := <<"subscribe">>, <<"calendar_id">> := CalendarId} ->
+      io:format("Subscribe to calendar: ~s~n", [CalendarId]),
+      NewSubs = [CalendarId | State#state.subscriptions],
+      Reply = jsx:encode(#{status => <<"subscribed">>, calendar_id => CalendarId}),
+      io:format("Sending reply: ~s~n", [Reply]),
+      {reply, {text, Reply}, State#state{subscriptions = NewSubs}};
+    #{<<"action">> := <<"ping">>} ->
+      {reply, {text, <<"{\"status\":\"pong\"}">>}, State};
+    Other ->
+      io:format("Unknown action: ~p~n", [Other]),
+      {ok, State}
+  catch
+    _:Error ->
+      io:format("Error parsing WebSocket message: ~p~n", [Error]),
+      {ok, State}
+  end;
+websocket_handle(_Frame, State) ->
+  {ok, State}.
+
+websocket_info({notification, Type, Data}, State) ->
+  case should_notify(Type, Data, State) of
+    true ->
+      Msg = jsx:encode(#{
+        type => Type,
+        data => Data,
+        timestamp => os:system_time(seconds)
+      }),
+      {reply, {text, Msg}, State};
+    false ->
+      {ok, State}
+  end;
+websocket_info(_Info, State) ->
+  {ok, State}.
+
+terminate(_Reason, _Req, _State) ->
+  pg:leave(eventhub_ws, self()),
+  ok.
+
+%% Проверка, нужно ли отправлять уведомление пользователю
+should_notify(calendar_update, #{calendar_id := CalId}, State) ->
+  lists:member(CalId, State#state.subscriptions);
+should_notify(booking_update, #{user_id := UserId}, State) ->
+  UserId =:= State#state.user_id;
+should_notify(event_update, #{calendar_id := CalId}, State) ->
+  lists:member(CalId, State#state.subscriptions);
+should_notify(_, _, _) ->
+  true.