138 lines
5.2 KiB
Erlang
138 lines
5.2 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @doc Пользовательский WebSocket-обработчик.
|
|
%%%
|
|
%%% Устанавливает WebSocket-соединение после проверки JWT-токена
|
|
%%% и подписывает пользователя на обновления календарей.
|
|
%%%
|
|
%%% Поддерживаемые действия:
|
|
%%% - subscribe – подписаться на обновления календаря
|
|
%%% - unsubscribe – отписаться от обновлений
|
|
%%% - ping – проверка соединения
|
|
%%% @end
|
|
%%%-------------------------------------------------------------------
|
|
-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()] % ← инициализация пустым списком
|
|
}).
|
|
|
|
%%%===================================================================
|
|
%%% cowboy_websocket callback
|
|
%%%===================================================================
|
|
|
|
-spec init(cowboy_req:req(), any()) ->
|
|
{ok, cowboy_req:req(), undefined} |
|
|
{cowboy_websocket, cowboy_req:req(), #state{}}.
|
|
init(Req, _Opts) ->
|
|
Qs = cowboy_req:parse_qs(Req),
|
|
case proplists:get_value(<<"token">>, Qs) of
|
|
undefined ->
|
|
Req1 = cowboy_req:reply(401, #{}, <<"Missing token">>, Req),
|
|
{ok, Req1, undefined};
|
|
Token ->
|
|
case logic_auth:verify_jwt(Token) of
|
|
{ok, UserId, _Role} ->
|
|
{cowboy_websocket, Req, #state{user_id = UserId}};
|
|
{error, _} ->
|
|
Req1 = cowboy_req:reply(401, #{}, <<"Invalid token">>, Req),
|
|
{ok, Req1, undefined}
|
|
end
|
|
end.
|
|
|
|
%%%===================================================================
|
|
%%% websocket callbacks
|
|
%%%===================================================================
|
|
|
|
-spec websocket_init(#state{}) -> {ok, #state{}}.
|
|
websocket_init(#state{user_id = UserId} = State) ->
|
|
pg:join(eventhub_ws, self()),
|
|
io:format("[WS] User ~s connected~n", [UserId]),
|
|
core_counters:inc(ws_connections),
|
|
{ok, State#state{subscriptions = []}}.
|
|
|
|
-spec websocket_handle(term(), #state{}) ->
|
|
{ok, #state{}} | {reply, {text, binary()}, #state{}}.
|
|
websocket_handle({text, Msg}, State) ->
|
|
io:format("[WS] Received: ~s~n", [Msg]),
|
|
try jsx:decode(Msg, [return_maps]) of
|
|
#{<<"action">> := <<"subscribe">>, <<"calendar_id">> := CalendarId} ->
|
|
handle_subscribe(CalendarId, State);
|
|
#{<<"action">> := <<"unsubscribe">>, <<"calendar_id">> := CalendarId} ->
|
|
handle_unsubscribe(CalendarId, State);
|
|
#{<<"action">> := <<"ping">>} ->
|
|
{reply, {text, <<"{\"status\":\"pong\"}">>}, State};
|
|
Other ->
|
|
io:format("[WS] Unknown action: ~p~n", [Other]),
|
|
{ok, State}
|
|
catch
|
|
_:Error ->
|
|
io:format("[WS] Error parsing message: ~p~n", [Error]),
|
|
{ok, State}
|
|
end;
|
|
websocket_handle(_Frame, State) ->
|
|
{ok, State}.
|
|
|
|
-spec websocket_info(term(), #state{}) ->
|
|
{ok, #state{}} | {reply, {text, binary()}, #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}.
|
|
|
|
-spec terminate(term(), cowboy_req:req(), #state{}) -> ok.
|
|
terminate(_Reason, _Req, #state{user_id = UserId}) ->
|
|
pg:leave(eventhub_ws, self()),
|
|
core_counters:dec(ws_connections),
|
|
io:format("[WS] User ~s disconnected~n", [UserId]),
|
|
ok.
|
|
|
|
%%%===================================================================
|
|
%%% Внутренние функции
|
|
%%%===================================================================
|
|
|
|
-spec handle_subscribe(binary(), #state{}) -> {reply, {text, binary()}, #state{}}.
|
|
handle_subscribe(CalendarId, #state{subscriptions = Subs} = State) ->
|
|
io:format("[WS] Subscribe to calendar: ~s~n", [CalendarId]),
|
|
NewSubs = case lists:member(CalendarId, Subs) of
|
|
true -> Subs;
|
|
false -> [CalendarId | Subs]
|
|
end,
|
|
Reply = jsx:encode(#{status => <<"subscribed">>, calendar_id => CalendarId}),
|
|
{reply, {text, Reply}, State#state{subscriptions = NewSubs}}.
|
|
|
|
-spec handle_unsubscribe(binary(), #state{}) -> {reply, {text, binary()}, #state{}}.
|
|
handle_unsubscribe(CalendarId, #state{subscriptions = Subs} = State) ->
|
|
io:format("[WS] Unsubscribe from calendar: ~s~n", [CalendarId]),
|
|
NewSubs = lists:delete(CalendarId, Subs),
|
|
Reply = jsx:encode(#{status => <<"unsubscribed">>, calendar_id => CalendarId}),
|
|
{reply, {text, Reply}, State#state{subscriptions = NewSubs}}.
|
|
|
|
-spec should_notify(atom(), map(), #state{}) -> boolean().
|
|
should_notify(calendar_update, #{calendar_id := CalId}, #state{subscriptions = Subs}) ->
|
|
lists:member(CalId, Subs);
|
|
should_notify(booking_update, #{user_id := UserId}, #state{user_id = UserId}) ->
|
|
true;
|
|
should_notify(booking_update, _, _) ->
|
|
false;
|
|
should_notify(event_update, #{calendar_id := CalId}, #state{subscriptions = Subs}) ->
|
|
lists:member(CalId, Subs);
|
|
should_notify(_, _, _) ->
|
|
true. |