feat(ft-api): support paging in S3 storage exporter

apps/emqx_ft/src/emqx_ft_storage_exporter_s3.erl

@@ -23,7 +23,7 @@
 -export([write/2]).
 -export([complete/2]).
 -export([discard/1]).
--export([list/1]).
+-export([list/2]).
 
 -export([
     start/1,
@@ -43,6 +43,10 @@
     filemeta => filemeta()
 }.
 
+-type query() :: emqx_ft_storage:query(cursor()).
+-type page(T) :: emqx_ft_storage:page(T, cursor()).
+-type cursor() :: iodata().
+
 -type export_st() :: #{
     pid := pid(),
     filemeta := filemeta(),
@@ -92,10 +96,10 @@ complete(#{pid := Pid} = _ExportSt, _Checksum) ->
 discard(#{pid := Pid} = _ExportSt) ->
     emqx_s3_uploader:abort(Pid).
 
--spec list(options()) ->
+-spec list(options(), query()) ->
-    {ok, [exportinfo()]} | {error, term()}.
+    {ok, page(exportinfo())} | {error, term()}.
-list(Options) ->
+list(Options, Query) ->
-    emqx_s3:with_client(?S3_PROFILE_ID, fun(Client) -> list(Client, Options) end).
+    emqx_s3:with_client(?S3_PROFILE_ID, fun(Client) -> list(Client, Options, Query) end).
 
 %%--------------------------------------------------------------------
 %% Exporter behaviour (lifecycle)
@@ -117,12 +121,11 @@ update(_OldOptions, NewOptions) ->
 %% Internal functions
 %% -------------------------------------------------------------------
 
-s3_key({ClientId, FileId} = _Transfer, #{name := Filename}) ->
+s3_key(Transfer, #{name := Filename}) ->
-    filename:join([
+    s3_prefix(Transfer) ++ "/" ++ Filename.
-        emqx_ft_fs_util:escape_filename(ClientId),
+
-        emqx_ft_fs_util:escape_filename(FileId),
+s3_prefix({ClientId, FileId} = _Transfer) ->
-        Filename
+    emqx_ft_fs_util:escape_filename(ClientId) ++ "/" ++ emqx_ft_fs_util:escape_filename(FileId).
-    ]).
 
 s3_headers({ClientId, FileId}, Filemeta) ->
     #{
@@ -137,54 +140,97 @@ s3_headers({ClientId, FileId}, Filemeta) ->
 s3_header_filemeta(Filemeta) ->
     emqx_utils_json:encode(emqx_ft:encode_filemeta(Filemeta), [force_utf8, uescape]).
 
-list(Client, Options) ->
+list(Client, _Options, #{transfer := Transfer}) ->
-    case list_key_info(Client, Options) of
+    case list_key_info(Client, [{prefix, s3_prefix(Transfer)}, {max_keys, ?S3_LIST_LIMIT}]) of
-        {ok, KeyInfos} ->
+        {ok, {Exports, _Marker}} ->
-            MaybeExportInfos = lists:map(
+            {ok, #{items => Exports}};
-                fun(KeyInfo) -> key_info_to_exportinfo(Client, KeyInfo, Options) end, KeyInfos
+        {error, _Reason} = Error ->
-            ),
+            Error
-            ExportInfos = [ExportInfo || {ok, ExportInfo} <- MaybeExportInfos],
+    end;
-            {ok, ExportInfos};
+list(Client, _Options, Query) ->
+    Limit = maps:get(limit, Query, undefined),
+    Marker = emqx_maybe:apply(fun decode_cursor/1, maps:get(cursor, Query, undefined)),
+    case list_pages(Client, Marker, Limit, []) of
+        {ok, {Exports, undefined}} ->
+            {ok, #{items => Exports}};
+        {ok, {Exports, NextMarker}} ->
+            {ok, #{items => Exports, cursor => encode_cursor(NextMarker)}};
         {error, _Reason} = Error ->
             Error
     end.
 
-list_key_info(Client, Options) ->
+list_pages(Client, Marker, Limit, Acc) ->
-    list_key_info(Client, Options, _Marker = [], _Acc = []).
+    MaxKeys = min(?S3_LIST_LIMIT, Limit),
+    ListOptions = [{marker, Marker} || Marker =/= undefined],
+    case list_key_info(Client, [{max_keys, MaxKeys} | ListOptions]) of
+        {ok, {Exports, NextMarker}} ->
+            list_accumulate(Client, Limit, NextMarker, [Exports | Acc]);
+        {error, _Reason} = Error ->
+            Error
+    end.
 
-list_key_info(Client, Options, Marker, Acc) ->
+list_accumulate(_Client, _Limit, undefined, Acc) ->
-    ListOptions = [{max_keys, ?S3_LIST_LIMIT}] ++ Marker,
+    {ok, {flatten_pages(Acc), undefined}};
+list_accumulate(Client, undefined, Marker, Acc) ->
+    list_pages(Client, Marker, undefined, Acc);
+list_accumulate(Client, Limit, Marker, Acc = [Exports | _]) ->
+    case Limit - length(Exports) of
+        0 ->
+            {ok, {flatten_pages(Acc), Marker}};
+        Left ->
+            list_pages(Client, Marker, Left, Acc)
+    end.
+
+flatten_pages(Pages) ->
+    lists:append(lists:reverse(Pages)).
+
+list_key_info(Client, ListOptions) ->
     case emqx_s3_client:list(Client, ListOptions) of
         {ok, Result} ->
             ?SLOG(debug, #{msg => "list_key_info", result => Result}),
             KeyInfos = proplists:get_value(contents, Result, []),
-            case proplists:get_value(is_truncated, Result, false) of
+            Exports = lists:filtermap(
-                true ->
+                fun(KeyInfo) -> key_info_to_exportinfo(Client, KeyInfo) end, KeyInfos
-                    NewMarker = next_marker(KeyInfos),
+            ),
-                    list_key_info(Client, Options, NewMarker, [KeyInfos | Acc]);
+            Marker =
-                false ->
+                case proplists:get_value(is_truncated, Result, false) of
-                    {ok, lists:append(lists:reverse([KeyInfos | Acc]))}
+                    true ->
-            end;
+                        next_marker(KeyInfos);
+                    false ->
+                        undefined
+                end,
+            {ok, {Exports, Marker}};
         {error, _Reason} = Error ->
             Error
     end.
 
-next_marker(KeyInfos) ->
+encode_cursor(Key) ->
-    [{marker, proplists:get_value(key, lists:last(KeyInfos))}].
+    unicode:characters_to_binary(Key).
 
-key_info_to_exportinfo(Client, KeyInfo, _Options) ->
+decode_cursor(Cursor) ->
+    case unicode:characters_to_list(Cursor) of
+        Key when is_list(Key) ->
+            Key;
+        _ ->
+            error({badarg, cursor})
+    end.
+
+next_marker(KeyInfos) ->
+    proplists:get_value(key, lists:last(KeyInfos)).
+
+key_info_to_exportinfo(Client, KeyInfo) ->
     Key = proplists:get_value(key, KeyInfo),
     case parse_transfer_and_name(Key) of
         {ok, {Transfer, Name}} ->
-            {ok, #{
+            {true, #{
                 transfer => Transfer,
                 name => unicode:characters_to_binary(Name),
                 uri => emqx_s3_client:uri(Client, Key),
                 timestamp => datetime_to_epoch_second(proplists:get_value(last_modified, KeyInfo)),
                 size => proplists:get_value(size, KeyInfo)
             }};
-        {error, _Reason} = Error ->
+        {error, _Reason} ->
-            Error
+            false
     end.
 
 -define(EPOCH_START, 62167219200).