%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*-
-module(mnesia_rocksdb_backup_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([ hybrid_backup_cf_and_disc/1
        , hybrid_backup_with_standalone/1
        , backup_rejects_nonempty_dir/1
        , backup_with_verify/1
        , restore_roundtrip_cf_and_disc/1
        , restore_roundtrip_with_standalone/1
        ]).

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

suite() ->
    [].

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

groups() ->
    [{all_tests, [sequence],
      [ hybrid_backup_cf_and_disc
      , hybrid_backup_with_standalone
      , backup_rejects_nonempty_dir
      , backup_with_verify
      , restore_roundtrip_cf_and_disc
      , restore_roundtrip_with_standalone
      ]}].

init_per_suite(Config) ->
    Config.

end_per_suite(_Config) ->
    ok.

init_per_group(_, Config) ->
    Config.

end_per_group(_, _Config) ->
    ok.

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

end_per_testcase(_TC, _Config) ->
    ok.

%%--------------------------------------------------------------------
%% Tests
%%--------------------------------------------------------------------

%% disc_copies (not ram_copies): with ram_overrides_dump=false, Mnesia
%% checkpoints prefer the dumped image for ram_copies, so pure RAM writes
%% are a poor stand-in for "persistent non-RocksDB tables".
hybrid_backup_cf_and_disc(Config) ->
    create_rdb_table(t, []),
    {atomic, ok} =
        mnesia:create_table(d, [{disc_copies, [node()]},
                                {attributes, [k, v]}]),
    ok = mnesia:wait_for_tables([t, d], 5000),

    [ok = mnesia:dirty_write({t, K, V}) || {K, V} <- [{a, 1}, {b, 2}, {c, 3}]],
    [ok = mnesia:dirty_write({d, K, V}) || {K, V} <- [{x, 10}, {y, 20}]],

    BackupDir = backup_dir(Config, "cf_disc"),
    ?assertEqual(ok, mnesia_rocksdb:backup(BackupDir)),

    assert_layout(BackupDir),
    Manifest = read_manifest(BackupDir),
    ct:log("MANIFEST = ~p", [Manifest]),

    ?assertEqual(1, maps:get(version, Manifest)),
    ?assert(lists:member(t, maps:get(rocksdb_tables, Manifest))),
    ?assertNot(lists:member(t, maps:get(mnesia_tables, Manifest))),
    ?assert(lists:member(schema, maps:get(mnesia_tables, Manifest))),
    ?assert(lists:member(d, maps:get(mnesia_tables, Manifest))),

    Dbs = maps:get(dbs, Manifest),
    ?assert(lists:any(fun(#{kind := admin}) -> true; (_) -> false end, Dbs)),

    assert_admin_backup_ok(BackupDir, Dbs),
    assert_mnesia_bup_has_table(BackupDir, d),
    assert_mnesia_bup_lacks_table_data(BackupDir, t),
    ok.

hybrid_backup_with_standalone(Config) ->
    create_rdb_table(t, []),
    create_rdb_table(ts, [{user_properties, [{rocksdb_standalone, true}]}]),
    {atomic, ok} =
        mnesia:create_table(d, [{disc_copies, [node()]},
                                {attributes, [k, v]}]),
    ok = mnesia:wait_for_tables([t, ts, d], 5000),

    ok = mnesia:dirty_write({t, 1, one}),
    ok = mnesia:dirty_write({ts, 2, two}),
    ok = mnesia:dirty_write({d, 3, three}),

    #{type := standalone} = mrdb:get_ref(ts),

    BackupDir = backup_dir(Config, "standalone"),
    ?assertEqual(ok, mnesia_rocksdb:backup(BackupDir)),

    Manifest = read_manifest(BackupDir),
    RdbTabs = maps:get(rocksdb_tables, Manifest),
    ?assert(lists:member(t, RdbTabs)),
    ?assert(lists:member(ts, RdbTabs)),
    ?assert(lists:member(d, maps:get(mnesia_tables, Manifest))),

    Dbs = maps:get(dbs, Manifest),
    ?assert(lists:any(fun(#{kind := standalone, name := ts}) -> true;
                         (_) -> false end, Dbs)),
    ?assert(lists:any(fun(#{kind := admin}) -> true; (_) -> false end, Dbs)),

    lists:foreach(
      fun(#{backup_relpath := Rel, backup_id := Id}) ->
              Abs = filename:join(BackupDir, Rel),
              {ok, BE} = rocksdb:open_backup_engine(Abs),
              try
                  ?assertEqual(ok, rocksdb:verify_backup(BE, Id))
              after
                  rocksdb:close_backup_engine(BE)
              end
      end, Dbs),
    ok.

backup_rejects_nonempty_dir(Config) ->
    create_rdb_table(t, []),
    ok = mnesia:wait_for_tables([t], 5000),
    ok = mnesia:dirty_write({t, 1, a}),

    BackupDir = backup_dir(Config, "nonempty"),
    ok = filelib:ensure_dir(filename:join(BackupDir, "noise")),
    case file:make_dir(BackupDir) of
        ok -> ok;
        {error, eexist} -> ok
    end,
    ok = file:write_file(filename:join(BackupDir, "noise"), <<"x">>),

    ?assertMatch({error, {not_empty, _}}, mnesia_rocksdb:backup(BackupDir)),
    ok.

backup_with_verify(Config) ->
    create_rdb_table(t, []),
    ok = mnesia:wait_for_tables([t], 5000),
    ok = mnesia:dirty_write({t, 1, a}),

    BackupDir = backup_dir(Config, "verify"),
    ?assertEqual(ok, mnesia_rocksdb:backup(BackupDir, #{verify => true})),
    assert_layout(BackupDir),
    ok.

restore_roundtrip_cf_and_disc(Config) ->
    create_rdb_table(t, []),
    {atomic, ok} =
        mnesia:create_table(d, [{disc_copies, [node()]},
                                {attributes, [k, v]}]),
    ok = mnesia:wait_for_tables([t, d], 5000),

    [ok = mnesia:dirty_write({t, K, V}) || {K, V} <- [{a, 1}, {b, 2}]],
    [ok = mnesia:dirty_write({d, K, V}) || {K, V} <- [{x, 10}, {y, 20}]],

    BackupDir = backup_dir(Config, "restore_cf"),
    ?assertEqual(ok, mnesia_rocksdb:backup(BackupDir)),

    %% Mutate after backup — restore must discard these changes.
    ok = mnesia:dirty_write({t, a, 999}),
    ok = mnesia:dirty_write({t, c, 3}),
    ok = mnesia:dirty_delete({d, x}),
    ok = mnesia:dirty_write({d, z, 30}),

    ?assertEqual(ok, mnesia_rocksdb:restore(BackupDir, #{start => true})),
    ?assertEqual(ok, mnesia:wait_for_tables([t, d], 10000)),

    ?assertEqual([{t, a, 1}, {t, b, 2}],
                 lists:sort(mnesia:dirty_match_object({t, '_', '_'}))),
    ?assertEqual([{d, x, 10}, {d, y, 20}],
                 lists:sort(mnesia:dirty_match_object({d, '_', '_'}))),
    ok.

restore_roundtrip_with_standalone(Config) ->
    create_rdb_table(t, []),
    create_rdb_table(ts, [{user_properties, [{rocksdb_standalone, true}]}]),
    {atomic, ok} =
        mnesia:create_table(d, [{disc_copies, [node()]},
                                {attributes, [k, v]}]),
    ok = mnesia:wait_for_tables([t, ts, d], 5000),

    ok = mnesia:dirty_write({t, 1, one}),
    ok = mnesia:dirty_write({ts, 2, two}),
    ok = mnesia:dirty_write({d, 3, three}),
    #{type := standalone} = mrdb:get_ref(ts),

    BackupDir = backup_dir(Config, "restore_sa"),
    ?assertEqual(ok, mnesia_rocksdb:backup(BackupDir)),

    ok = mnesia:dirty_write({t, 1, changed}),
    ok = mnesia:dirty_write({ts, 2, changed}),
    ok = mnesia:dirty_write({d, 3, changed}),

    ?assertEqual(ok, mnesia_rocksdb:restore(BackupDir, #{start => true})),
    ?assertEqual(ok, mnesia:wait_for_tables([t, ts, d], 10000)),

    ?assertEqual([{t, 1, one}], mnesia:dirty_read(t, 1)),
    ?assertEqual([{ts, 2, two}], mnesia:dirty_read(ts, 2)),
    ?assertEqual([{d, 3, three}], mnesia:dirty_read(d, 3)),
    #{type := standalone} = mrdb:get_ref(ts),
    ok.

%%--------------------------------------------------------------------
%% Helpers
%%--------------------------------------------------------------------

create_rdb_table(Name, Extra) ->
    {atomic, ok} =
        mnesia:create_table(Name, [{rdb, [node()]},
                                   {attributes, [k, v]} | Extra]),
    ok.

backup_dir(Config, Name) ->
    filename:join(?config(priv_dir, Config), "backup_" ++ Name).

assert_layout(BackupDir) ->
    ?assert(filelib:is_regular(filename:join(BackupDir, "MANIFEST"))),
    ?assert(filelib:is_regular(filename:join(BackupDir, "mnesia.BUP"))),
    ?assert(filelib:is_dir(filename:join(BackupDir, "rocksdb"))).

read_manifest(BackupDir) ->
    Path = filename:join(BackupDir, "MANIFEST"),
    {ok, [Manifest]} = file:consult(Path),
    Manifest.

assert_admin_backup_ok(BackupDir, Dbs) ->
    [Admin | _] = [D || #{kind := admin} = D <- Dbs],
    #{backup_relpath := Rel, backup_id := Id} = Admin,
    Abs = filename:join(BackupDir, Rel),
    {ok, BE} = rocksdb:open_backup_engine(Abs),
    try
        Infos = case rocksdb:get_backup_info(BE) of
                    {ok, L} -> L;
                    L when is_list(L) -> L
                end,
        ?assertNotEqual([], Infos),
        ?assertEqual(ok, rocksdb:verify_backup(BE, Id))
    after
        rocksdb:close_backup_engine(BE)
    end.

assert_mnesia_bup_has_table(BackupDir, Tab) ->
    Bup = filename:join(BackupDir, "mnesia.BUP"),
    %% Fun is applied per backup item (not per batch).
    {ok, Found} =
        mnesia:traverse_backup(
          Bup, mnesia_backup, read_only, read_only,
          fun(Rec, _Acc) when is_tuple(Rec), element(1, Rec) =:= Tab ->
                  {[Rec], true};
             (Rec, Acc) ->
                  {[Rec], Acc}
          end, false),
    ?assertEqual(true, Found).

assert_mnesia_bup_lacks_table_data(BackupDir, Tab) ->
    Bup = filename:join(BackupDir, "mnesia.BUP"),
    {ok, Found} =
        mnesia:traverse_backup(
          Bup, mnesia_backup, read_only, read_only,
          fun({schema, _, _} = Rec, Acc) ->
                  {[Rec], Acc};
             (Rec, _Acc) when is_tuple(Rec), element(1, Rec) =:= Tab ->
                  {[Rec], true};
             (Rec, Acc) ->
                  {[Rec], Acc}
          end, false),
    %% Data records for Tab should not appear; schema may still mention Tab.
    ?assertEqual(false, Found).