fix efka_client

include/iot_tables.hrl

@@ -0,0 +1,16 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2026, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 24. 4月 2026 16:54
+%%%-------------------------------------------------------------------
+-author("anlicheng").
+
+-record(efka_client, {
+    uuid :: binary(),
+    token_hash :: binary(),
+    salt :: binary(),
+    timestamp = 0 :: integer()
+}).

src/iot_app.erl

@@ -34,6 +34,7 @@ start_mnesia() ->
     ok = ensure_mnesia_schema(),
     %% 启动数据库
     ok = mnesia:start(),
+    ok = ensure_mnesia_tables(),
     Tables = mnesia:system_info(tables),
     logger:debug("[iot_app] tables: ~p", [Tables]),
     %% 创建数据库表
@@ -110,3 +111,19 @@ ensure_mnesia_schema() ->
                     throw({init_schema, Error})
             end
     end.
+
+-spec ensure_mnesia_tables() -> ok.
+ensure_mnesia_tables() ->
+    ok = ensure_efka_client_table(),
+    ok = mnesia:wait_for_tables([efka_client], 5000),
+    ok.
+
+-spec ensure_efka_client_table() -> ok.
+ensure_efka_client_table() ->
+    case efka_client_store:ensure_table() of
+        ok ->
+            ok;
+        {error, Reason} ->
+            logger:debug("[iot_app] create efka_client table failed with error: ~p", [Reason]),
+            throw({init_tables, {efka_client, Reason}})
+    end.

src/mnesia/efka_client_store.erl

@@ -0,0 +1,121 @@
+%%%-------------------------------------------------------------------
+%%% @doc efka_client table store based on mnesia.
+%%%-------------------------------------------------------------------
+-module(efka_client_store).
+
+-include("iot_tables.hrl").
+
+-export([ensure_table/0, register/2, get/1, update/1, delete/1, list/0]).
+-export([auth/2]).
+
+-define(TOKEN_HASH_ITERATIONS, 32).
+-define(TOKEN_HASH_BYTES, 32).
+
+-spec ensure_table() -> ok | {error, term()}.
+ensure_table() ->
+    case mnesia:create_table(efka_client, [
+        {attributes, record_info(fields, efka_client)},
+        {type, set},
+        {record_name, efka_client},
+        {disc_copies, [node()]}
+    ]) of
+        {atomic, ok} ->
+            ok;
+        {aborted, {already_exists, efka_client}} ->
+            ok;
+        {aborted, Reason} ->
+            {error, Reason}
+    end.
+
+-spec auth(UUID :: binary(), Token :: binary()) -> boolean().
+auth(UUID, Token) when is_binary(UUID), UUID =/= <<>>, is_binary(Token) ->
+    case mnesia:dirty_read(efka_client, UUID) of
+        [] ->
+            false;
+        [#efka_client{token_hash = TokenHash, salt = Salt}] ->
+            hash_token(Token, Salt) =:= TokenHash
+    end.
+
+-spec register(UUID :: binary(), Token :: binary()) -> ok | {error, term()}.
+register(UUID, Token) when is_binary(UUID), UUID =/= <<>>, is_binary(Token) ->
+    case mnesia:transaction(fun() ->
+        case mnesia:read(efka_client, UUID, write) of
+            [] ->
+                %% 1. 生成 16 字节随机盐
+                Salt = crypto:strong_rand_bytes(16),
+                TokenHash = hash_token(Token, Salt),
+                ok = mnesia:write(#efka_client{uuid = UUID, token_hash = TokenHash, salt = Salt, timestamp = iot_util:timestamp()}),
+                ok;
+            [_] ->
+                {error, already_exists}
+        end
+                            end) of
+        {atomic, ok} ->
+            ok;
+        {atomic, {error, Reason}} ->
+            {error, Reason};
+        {aborted, Reason} ->
+            {error, Reason}
+    end.
+
+-spec get(binary()) -> {ok, #efka_client{}} | not_found | {error, term()}.
+get(UUID) when is_binary(UUID), UUID =/= <<>> ->
+    case mnesia:transaction(fun() ->
+        case mnesia:read(efka_client, UUID, read) of
+            [Client = #efka_client{}] ->
+                {ok, Client};
+            [] ->
+                not_found
+        end
+    end) of
+        {atomic, Result} ->
+            Result;
+        {aborted, Reason} ->
+            {error, Reason}
+    end.
+
+-spec update(#efka_client{}) -> ok | {error, term()}.
+update(Client = #efka_client{uuid = UUID}) when is_binary(UUID), UUID =/= <<>> ->
+    case mnesia:transaction(fun() ->
+        case mnesia:read(efka_client, UUID, write) of
+            [_] ->
+                ok = mnesia:write(Client),
+                ok;
+            [] ->
+                {error, not_found}
+        end
+    end) of
+        {atomic, ok} ->
+            ok;
+        {atomic, {error, Reason}} ->
+            {error, Reason};
+        {aborted, Reason} ->
+            {error, Reason}
+    end.
+
+-spec delete(binary()) -> ok | {error, term()}.
+delete(UUID) when is_binary(UUID), UUID =/= <<>> ->
+    case mnesia:transaction(fun() ->
+        ok = mnesia:delete({efka_client, UUID}),
+        ok
+    end) of
+        {atomic, ok} ->
+            ok;
+        {aborted, Reason} ->
+            {error, Reason}
+    end.
+
+-spec list() -> {ok, [#efka_client{}]} | {error, term()}.
+list() ->
+    case mnesia:transaction(fun() ->
+        mnesia:foldl(fun(Client = #efka_client{}, Acc) -> [Client | Acc] end, [], efka_client)
+    end) of
+        {atomic, Clients} ->
+            {ok, lists:reverse(Clients)};
+        {aborted, Reason} ->
+            {error, Reason}
+    end.
+
+-spec hash_token(binary(), binary()) -> binary().
+hash_token(Token, Salt) when is_binary(Token), is_binary(Salt) ->
+    crypto:pbkdf2_hmac(sha256, Token, Salt, ?TOKEN_HASH_ITERATIONS, ?TOKEN_HASH_BYTES).

src/transport/tcp/ssl_channel.erl

@@ -150,7 +150,7 @@ handle_info({stop, Reason}, State) ->
 %% 主机进程挂掉的时候，需要关闭掉链接
 handle_info({'DOWN', _, process, HostPid, Reason}, State = #state{uuid = UUID, host_pid = HostPid}) ->
     logger:debug("[ws_channel] uuid: ~p, channel will close because host exited with reason: ~p", [UUID, Reason]),
-    {stop, State};
+    {stop, Reason, State};
 
 handle_info({ssl, Socket, <<?FRAME_REQUEST, FrameBin/binary>>},
             State = #state{transport = Transport, socket = Socket}) ->
@@ -228,7 +228,7 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId,
             case iot_api_client:get_host_by_uuid(UUID) of
                 undefined ->
                     logger:warning("[ws_channel] uuid: ~p, user: ~p, host not found", [UUID, Username]),
-                    {stop, State};
+                    {stop, normal, State};
                 {ok, _} ->
                     %% 尝试启动主机的服务进程
                     {ok, HostPid} = iot_host_sup:ensured_host_started(UUID),
@@ -245,20 +245,20 @@ handle_request_frame(#'RequestFrame'{packet_id = PacketId,
                         {error, Reason} when is_binary(Reason) ->
                             send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = Reason}}),
                             logger:debug("[ws_channel] uuid: ~p, attach channel get error: ~p, stop channel", [UUID, Reason]),
-                            {stop, State}
+                            {stop, Reason, State}
                     end
             end;
         {error, Reason} ->
             send_reply_frame(Transport, Socket, PacketId, {error, #'ReplyFrame.Error'{code = 2, message = Reason}}),
             logger:warning("[ws_channel] uuid: ~p, user: ~p, auth failed, reason: ~p", [UUID, Username, Reason]),
-            {stop, State}
+            {stop, Reason, State}
     end;
 handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = {container_request, ContainerRequest}}, _Transport, _Socket, State) ->
     logger:warning("[ws_channel] unsupported request message type: container_request, packet_id: ~p, request: ~p", [PacketId, ContainerRequest]),
-    {stop, State};
+    {stop, normal, State};
 handle_request_frame(#'RequestFrame'{packet_id = PacketId, body = undefined}, _Transport, _Socket, State) ->
     logger:warning("[ws_channel] empty request frame, packet_id: ~p", [PacketId]),
-    {stop, State}.
+    {stop, normal, State}.
 
 -spec handle_cast_frame(message_pb:'CastFrame'(), undefined | pid(), #state{}) ->
     {noreply, #state{}}.