-module(mnesia_rocksdb_migration_SUITE).

-export([
         all/0
       , suite/0
       , groups/0
       , init_per_suite/1
       , end_per_suite/1
       , init_per_group/2
       , end_per_group/2
       , init_per_testcase/2
       , end_per_testcase/2
        ]).

-export([
         manual_migration/1
       , migrate_with_encoding_change/1
       , auto_migration/1
       , online_encoding_migration/1
       , online_encoding_migration_restart/1
       , online_encoding_migration_interrupt/1
       , change_table_type_sync/1
       , change_table_type_async/1
       , change_table_type_schema_only/1
       , change_table_type_no_change/1
        ]).

-include_lib("common_test/include/ct.hrl").

-define(TABS_CREATED, tables_created).

suite() ->
    [].

all() ->
    [{group, all_tests}].

groups() ->
    [
     {all_tests, [sequence], [ manual_migration
                             , migrate_with_encoding_change
                             , online_encoding_migration
                             , online_encoding_migration_restart
                             , online_encoding_migration_interrupt
                             , change_table_type_sync
                             , change_table_type_async
                             , change_table_type_schema_only
                             , change_table_type_no_change ]}
    ].

init_per_suite(Config) ->
    Config.

end_per_suite(_Config) ->
    ok.

init_per_group(_, Config) ->
    Config.

end_per_group(_, _Config) ->
    ok.

init_per_testcase(_, Config) ->
    mnesia:stop(),
    ok = mnesia_rocksdb_tlib:start_mnesia(reset),
    Config.
%%    create_migrateable_db(Config).

end_per_testcase(_, _Config) ->
    ok.

manual_migration(Config) ->
    tr_ct:with_trace(fun manual_migration_/1, Config, tr_opts()).

manual_migration_(Config) ->
    create_migrateable_db(Config),
    Tabs = tables(),
    ct:log("Analyze (before): ~p", [analyze_tabs(Tabs)]),
    Res = mnesia_rocksdb_admin:migrate_standalone(rdb, Tabs),
    ct:log("migrate_standalone(rdb, ~p) -> ~p", [Tabs, Res]),
    AnalyzeRes = analyze_tabs(Tabs),
    ct:log("AnalyzeRes = ~p", [AnalyzeRes]),
    MigRes = mnesia_rocksdb_admin:migrate_standalone(rdb, Tabs),
    ct:log("MigRes = ~p", [MigRes]),
    AnalyzeRes2 = analyze_tabs(Tabs),
    ct:log("AnalyzeRes2 = ~p", [AnalyzeRes2]),
    ct:log("Admin State = ~p", [sys:get_state(mnesia_rocksdb_admin)]),
    ok.

migrate_with_encoding_change(_Config) ->
    ok = create_tab(t, [{user_properties, [{mrdb_encoding, {sext,{object,term}}},
                                           {rocksdb_standalone, true}]},
                        {index,[val]}
                       ]),
    mrdb:insert(t, {t, <<"1">>, <<"a">>}),
    mrdb:insert(t, {t, <<"2">>, <<"b">>}),
    TRef = mrdb:get_ref(t),
    {ok, V1} = mrdb:rdb_get(TRef, sext:encode(<<"1">>), []),
    {ok, V2} = mrdb:rdb_get(TRef, sext:encode(<<"2">>), []),
    {t,[],<<"a">>} = binary_to_term(V1),
    {t,[],<<"b">>} = binary_to_term(V2),
    Opts = #{encoding => {raw, raw}},
    MigRes = mnesia_rocksdb_admin:migrate_standalone(rdb, [{t, Opts}]),
    ct:log("MigRes (t) = ~p", [MigRes]),
    %%
    %% Ensure that metadata reflect the migrated table
    %% (now a column family, and the rocksdb_standalone prop gone)
    %%
    TRef1 = mrdb:get_ref(t),
    ct:log("TRef1(t) = ~p", [TRef1]),
    #{type := column_family,
      properties := #{user_properties := UPs}} = TRef1,
    error = maps:find(rocksdb_standalone, UPs),
    UPsR = lists:sort(maps:values(UPs)),
    UPsM = lists:sort(mnesia:table_info(t, user_properties)),
    {UPsR,UPsM} = {UPsM,UPsR},
    ct:log("user properties (t): ~p", [UPsM]),
    [{<<"2">>, <<"b">>},
     {<<"1">>, <<"a">>}] = mrdb:rdb_fold(
                             t, fun(K,V,A) -> [{K,V}|A] end, [], <<>>),
    ct:log("All data present in new column family", []),
    ct:log("Contents of mnesia dir: ~p",
           [ok(file:list_dir(mnesia:system_info(directory)))]),
    ct:log("mnesia stopped", []),
    mnesia:stop(),
    mnesia:start(),
    ct:log("mnesia started", []),
    mnesia:info(),
    ok = mnesia:wait_for_tables([t], 3000),
    ct:log("tables loaded", []),
    [{t,<<"1">>,<<"a">>},
     {t,<<"2">>,<<"b">>}] = mrdb:select(
                              t, [{'_',[],['$_']}]),
    [{<<"2">>,<<"b">>},
     {<<"1">>,<<"a">>}] = mrdb:rdb_fold(
                            t, fun(K,V,A) -> [{K,V}|A] end, [], <<>>),
    ok.

auto_migration(_Config) ->
    ok.

%% Online encoding migration: term keys -> sext keys while dual-writing.
online_encoding_migration(_Config) ->
    ok = create_tab(enc, [{attributes, [k, v]}]),
    %% Default set encoding is {term, {value, term}} for 2-attr tables.
    lists:foreach(
      fun(I) ->
              mrdb:insert(enc, {enc, I, I * 10})
      end, lists:seq(1, 50)),
    50 = length(mrdb:select(enc, [{'_', [], ['$_']}])),
    Ref0 = mrdb:get_ref(enc),
    ct:log("Before migration ref: ~p", [maps:with([encoding, type, semantics], Ref0)]),
    {term, _} = maps:get(encoding, Ref0),
    %% Concurrent writes during migration
    ok = mnesia_rocksdb_admin:migrate_encoding(
           rdb, enc, {sext, {value, term}}),
    %% Dual-write should be active
    #{migration := _} = mrdb:get_ref(enc),
    mrdb:insert(enc, {enc, 100, 1000}),
    mrdb:delete(enc, 5),
    %% Wait for copy_done
    ok = wait_copy_done(enc, 50),
    ct:log("Status after copy: ~p", [mnesia_rocksdb_admin:migration_status(enc)]),
    ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc),
    Ref1 = mrdb:get_ref(enc),
    ct:log("After finalize ref: ~p", [maps:with([encoding, type], Ref1)]),
    {sext, _} = maps:get(encoding, Ref1),
    false = maps:is_key(migration, Ref1),
    Objs = lists:sort(mrdb:select(enc, [{'_', [], ['$_']}])),
    %% 50 - 1 deleted + 1 inserted = 50
    50 = length(Objs),
    false = lists:keymember(5, 2, Objs),
    true = lists:keymember(100, 2, Objs),
    {enc, 100, 1000} = lists:keyfind(100, 2, Objs),
    %% sext-encoded key is readable via rdb_get
    {ok, _} = mrdb:rdb_get(Ref1, sext:encode(100), []),
    ok.

wait_copy_done(_Tab, 0) ->
    {error, timeout};
wait_copy_done(Tab, N) ->
    case mnesia_rocksdb_admin:migration_status(Tab) of
        #{phase := copy_done} ->
            ok;
        _ ->
            timer:sleep(50),
            wait_copy_done(Tab, N - 1)
    end.

%% Complete migration, restart mnesia, verify encoding + data survive.
online_encoding_migration_restart(_Config) ->
    ok = create_tab(enc_r, [{attributes, [k, v]}]),
    [mrdb:insert(enc_r, {enc_r, I, I * 3}) || I <- lists:seq(1, 30)],
    ok = mnesia_rocksdb_admin:migrate_encoding(
           rdb, enc_r, {sext, {value, term}}),
    ok = wait_copy_done(enc_r, 100),
    ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc_r),
    {sext, _} = maps:get(encoding, mrdb:get_ref(enc_r)),
    Before = lists:sort(mrdb:select(enc_r, [{'_', [], ['$_']}])),
    30 = length(Before),
    %% Close and reopen DB (same VM: stop/start mnesia + wait for tables).
    stopped = mnesia:stop(),
    ok = mnesia:start(),
    ok = mnesia:wait_for_tables([enc_r], 10000),
    Ref = mrdb:get_ref(enc_r),
    ct:log("After restart ref: ~p",
           [maps:with([encoding, cf_gen, type], Ref)]),
    {sext, _} = maps:get(encoding, Ref),
    false = maps:is_key(migration, Ref),
    After = lists:sort(mrdb:select(enc_r, [{'_', [], ['$_']}])),
    Before = After,
    {ok, _} = mrdb:rdb_get(Ref, sext:encode(1), []),
    ok.

%% Stop mid-migration (after dual-write is up), reopen, finish copy + finalize.
online_encoding_migration_interrupt(_Config) ->
    ok = create_tab(enc_i, [{attributes, [k, v]}]),
    [mrdb:insert(enc_i, {enc_i, I, I}) || I <- lists:seq(1, 100)],
    ok = mnesia_rocksdb_admin:migrate_encoding(
           rdb, enc_i, {sext, {value, term}}),
    %% Dual-write active; do not wait for copy_done — interrupt promptly.
    #{migration := _} = mrdb:get_ref(enc_i),
    mrdb:insert(enc_i, {enc_i, 200, 200}),
    mrdb:delete(enc_i, 10),
    stopped = mnesia:stop(),
    ok = mnesia:start(),
    ok = mnesia:wait_for_tables([enc_i], 10000),
    %% Migration should be re-armed (dual-write) or already copy_done.
    case mnesia_rocksdb_admin:migration_status(enc_i) of
        #{phase := Phase} when Phase =:= copying; Phase =:= copy_done ->
            ok;
        Other ->
            ct:fail({expected_migration_after_restart, Other})
    end,
    %% Writes during resumed dual-write
    mrdb:insert(enc_i, {enc_i, 201, 201}),
    ok = wait_copy_done(enc_i, 200),
    ok = mnesia_rocksdb_admin:finalize_migration(rdb, enc_i),
    {sext, _} = maps:get(encoding, mrdb:get_ref(enc_i)),
    false = maps:is_key(migration, mrdb:get_ref(enc_i)),
    Objs = lists:sort(mrdb:select(enc_i, [{'_', [], ['$_']}])),
    false = lists:keymember(10, 2, Objs),
    true = lists:keymember(200, 2, Objs),
    true = lists:keymember(201, 2, Objs),
    %% 100 - 1 delete + 2 inserts = 101
    101 = length(Objs),
    ok.

%% =====================================================================
%% change_table_type/3 and /4
%% =====================================================================

%% change_table_type/3 defaults to wait=true: copy + finalize in one call.
%% set (term keys) -> ordered_set implies default sext key encoding migration.
change_table_type_sync(_Config) ->
    ok = create_tab(ctt_s, [{attributes, [k, v]}]),
    set = mnesia:table_info(ctt_s, type),
    [mrdb:insert(ctt_s, {ctt_s, I, I * 2}) || I <- lists:seq(1, 40)],
    Ref0 = mrdb:get_ref(ctt_s),
    {term, _} = maps:get(encoding, Ref0),
    set = maps:get(type, maps:get(properties, Ref0)),
    ok = mnesia_rocksdb_admin:change_table_type(
           rdb, ctt_s, #{type => ordered_set, report => false}),
    idle = mnesia_rocksdb_admin:migration_status(ctt_s),
    Ref1 = mrdb:get_ref(ctt_s),
    false = maps:is_key(migration, Ref1),
    {sext, _} = maps:get(encoding, Ref1),
    ordered_set = maps:get(type, maps:get(properties, Ref1)),
    ordered_set = maps:get(semantics, Ref1),
    ordered_set = mnesia:table_info(ctt_s, type),
    assert_mrdb_encoding_up(ctt_s, {sext, {value, term}}),
    Objs = lists:sort(mrdb:select(ctt_s, [{'_', [], ['$_']}])),
    40 = length(Objs),
    {ok, _} = mrdb:rdb_get(Ref1, sext:encode(1), []),
    %% Writes after change use ordered_set / sext
    ok = mrdb:insert(ctt_s, {ctt_s, 100, 200}),
    {ctt_s, 100, 200} = lists:keyfind(100, 2, mrdb:select(ctt_s, [{'_', [], ['$_']}])),
    {ok, _} = mrdb:rdb_get(mrdb:get_ref(ctt_s), sext:encode(100), []),
    ok.

%% change_table_type/4 defaults to wait=false: start only; finalize manually.
change_table_type_async(_Config) ->
    ok = create_tab(ctt_a, [{attributes, [k, v]}]),
    [mrdb:insert(ctt_a, {ctt_a, I, I}) || I <- lists:seq(1, 25)],
    Self = self(),
    ok = mnesia_rocksdb_admin:change_table_type(
           rdb, ctt_a, #{type => ordered_set, encoding => {sext, {value, term}}},
           Self),
    %% Dual-write should be on immediately; type not cut over yet.
    #{migration := _} = mrdb:get_ref(ctt_a),
    set = mnesia:table_info(ctt_a, type),
    {term, _} = maps:get(encoding, mrdb:get_ref(ctt_a)),
    mrdb:insert(ctt_a, {ctt_a, 50, 50}),
    mrdb:delete(ctt_a, 3),
    ok = wait_copy_done(ctt_a, 100),
    ok = mnesia_rocksdb_admin:finalize_migration(rdb, ctt_a),
    idle = mnesia_rocksdb_admin:migration_status(ctt_a),
    Ref1 = mrdb:get_ref(ctt_a),
    false = maps:is_key(migration, Ref1),
    {sext, _} = maps:get(encoding, Ref1),
    ordered_set = mnesia:table_info(ctt_a, type),
    ordered_set = maps:get(type, maps:get(properties, Ref1)),
    Objs = lists:sort(mrdb:select(ctt_a, [{'_', [], ['$_']}])),
    false = lists:keymember(3, 2, Objs),
    true = lists:keymember(50, 2, Objs),
    %% 25 - 1 delete + 1 insert = 25
    25 = length(Objs),
    {ok, _} = mrdb:rdb_get(Ref1, sext:encode(50), []),
    ok.

%% Type change only when encoding already matches the default for the new type
%% (set with sext -> ordered_set with same sext): schema/metadata, no CF copy.
change_table_type_schema_only(_Config) ->
    ok = create_tab(ctt_so,
                    [{attributes, [k, v]},
                     {user_properties,
                      [{mrdb_encoding, {sext, {value, term}}}]}]),
    set = mnesia:table_info(ctt_so, type),
    Ref0 = mrdb:get_ref(ctt_so),
    {sext, _} = maps:get(encoding, Ref0),
    CfGen0 = maps:get(cf_gen, Ref0, 0),
    [mrdb:insert(ctt_so, {ctt_so, I, I}) || I <- lists:seq(1, 10)],
    ok = mnesia_rocksdb_admin:change_table_type(
           rdb, ctt_so, #{type => ordered_set, report => false}),
    idle = mnesia_rocksdb_admin:migration_status(ctt_so),
    Ref1 = mrdb:get_ref(ctt_so),
    false = maps:is_key(migration, Ref1),
    {sext, _} = maps:get(encoding, Ref1),
    CfGen0 = maps:get(cf_gen, Ref1, 0),
    ordered_set = mnesia:table_info(ctt_so, type),
    ordered_set = maps:get(type, maps:get(properties, Ref1)),
    ordered_set = maps:get(semantics, Ref1),
    10 = length(mrdb:select(ctt_so, [{'_', [], ['$_']}])),
    {ok, _} = mrdb:rdb_get(Ref1, sext:encode(1), []),
    ok.

%% Idempotent / no-op errors.
change_table_type_no_change(_Config) ->
    ok = create_tab(ctt_nc, [{attributes, [k, v]}]),
    set = mnesia:table_info(ctt_nc, type),
    {error, no_change} =
        mnesia_rocksdb_admin:change_table_type(
          rdb, ctt_nc, #{type => set, report => false}),
    %% Same encoding as live ref
    {term, ValEnc} = maps:get(encoding, mrdb:get_ref(ctt_nc)),
    {error, no_change} =
        mnesia_rocksdb_admin:change_table_type(
          rdb, ctt_nc, #{encoding => {term, ValEnc}, report => false}),
    ok.

assert_mrdb_encoding_up(Tab, Enc) ->
    UPs = mnesia:table_info(Tab, user_properties),
    {mrdb_encoding, Enc} = lists:keyfind(mrdb_encoding, 1, UPs),
    #{properties := #{user_properties := UPMap}} = mrdb:get_ref(Tab),
    {mrdb_encoding, Enc} = maps:get(mrdb_encoding, UPMap),
    ok.

ok({ok, Value}) -> Value.

tr_opts() ->
    #{ patterns => [ {mnesia_rocksdb_admin, '_', []}
                   , {mnesia_rocksdb_lib, '_', []}
                   , {rocksdb, '_', x} | trace_exports(mrdb, x) ] }.

trace_exports(M, Pat) ->
    Fs = M:module_info(exports),
    [{M, F, A, Pat} || {F, A} <- Fs].

tables() ->
    [a].

create_migrateable_db(Config) ->
    Os = [{user_properties, [{rocksdb_standalone, true}]}],
    TabNames = tables(),
    Tabs = [{T, Os} || T <- TabNames],
    create_tabs(Tabs, Config),
    verify_tabs_are_standalone(TabNames),
    fill_tabs(TabNames),
    Config.

fill_tabs(Tabs) ->
    %% Fill with more than 300, since that's the currently hard-coded chunk size
    lists:foreach(fun(Tab) ->
                          [mrdb:insert(Tab, {Tab, X, a}) || X <- lists:seq(1,500)]
                  end, Tabs).

create_tabs(Tabs, Config) ->
    Res = lists:map(fun create_tab/1, Tabs),
    tr_ct:trace_checkpoint(?TABS_CREATED, Config),
    Res.

create_tab({T, Opts}) ->
    create_tab(T, Opts).

create_tab(T, Opts) ->
    {atomic, ok} = mnesia:create_table(T, [{rdb, [node()]} | Opts]),
    ok.

verify_tabs_are_standalone(Tabs) ->
    case analyze_tabs(Tabs) of
        {_, []} ->
            ok;
        {[], NotSA} ->
            error({not_standalone, NotSA})
    end.

analyze_tabs(Tabs) ->
    Dir = mnesia:system_info(directory),
    Files = filelib:wildcard(filename:join(Dir, "*-_tab.extrdb")),
    ct:log("Files = ~p", [Files]),
    TabNames = lists:map(
                 fun(F) ->
                         {match,[TStr]} =
                            re:run(F, "^.+/([^/]+)-_tab\\.extrdb$",
                                   [{capture, [1], list}]),
                         list_to_existing_atom(TStr)
                 end, Files),
    ct:log("TabNames = ~p", [TabNames]),
    NotSA = Tabs -- TabNames,
    {TabNames -- NotSA, NotSA}.