From 6c0775ee5e5f493ebe54f7afc84e1cd5b4fca8a4 Mon Sep 17 00:00:00 2001 From: Aleksey Sabilin Date: Mon, 3 Aug 2026 22:57:41 +0300 Subject: [PATCH] feat(infra): add TTL cleanup worker for append-only tables Periodically prune old read notifications, revoked/expired auth sessions, and admin_audit in chunked transactions; wire under eventhub config. --- src/config/sys.config | 4 +- src/infra/infra_cleanup.erl | 204 ++++++++++++++++++++++++++++++ src/infra/infra_mnesia.erl | 1 + src/infra/infra_sup.erl | 10 +- test/unit/infra_cleanup_tests.erl | 132 +++++++++++++++++++ 5 files changed, 348 insertions(+), 3 deletions(-) create mode 100644 src/infra/infra_cleanup.erl create mode 100644 test/unit/infra_cleanup_tests.erl diff --git a/src/config/sys.config b/src/config/sys.config index c8763c3..23cbed6 100644 --- a/src/config/sys.config +++ b/src/config/sys.config @@ -5,7 +5,9 @@ {admin_http_port, "${ADMIN_HTTP_PORT:-8445}"}, {admin_ws_port, "${ADMIN_WS_PORT:-8446}"}, {jwt_secret, <<"${JWT_SECRET:-change_me_in_production}">>}, - {admin_jwt_secret, <<"${ADMIN_JWT_SECRET:-change_me_in_production}">>} + {admin_jwt_secret, <<"${ADMIN_JWT_SECRET:-change_me_in_production}">>}, + %% Interval between TTL cleanup runs (infra_cleanup), default 1 hour + {cleanup_interval_ms, 3600000} ]}, {kernel, [ {logger_level, info}, diff --git a/src/infra/infra_cleanup.erl b/src/infra/infra_cleanup.erl new file mode 100644 index 0000000..5aa9a0a --- /dev/null +++ b/src/infra/infra_cleanup.erl @@ -0,0 +1,204 @@ +%% =================================================================== +%% EventHub – периодическая очистка append-only таблиц (TTL cleanup) +%% =================================================================== +%% gen_server, который раз в N миллисекунд (по умолчанию 1 час) запускает +%% фоновый процесс очистки устаревших записей: +%% • notification — прочитанные старше 30 дней +%% • auth_session — отозванные (revoked) или истёкшие старше 7 дней +%% • admin_audit — записи старше 90 дней +%% +%% Deletes run in bounded chunks so a single tick does not hold long +%% Mnesia locks on large tables. +%% =================================================================== +-module(infra_cleanup). +-behaviour(gen_server). + +-include("records.hrl"). +-include_lib("stdlib/include/ms_transform.hrl"). + +%% API +-export([start_link/0, run_once/0, cutoff_datetime/1]). + +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +%% TTL thresholds in days +-define(NOTIFICATION_TTL_DAYS, 30). +-define(AUTH_SESSION_TTL_DAYS, 7). +-define(ADMIN_AUDIT_TTL_DAYS, 90). + +%% Default cleanup interval: 1 hour +-define(DEFAULT_INTERVAL_MS, 3600000). + +%% Chunked delete limits +-define(CHUNK_SIZE, 500). +-define(MAX_CHUNKS_PER_TICK, 20). + +-record(state, { + interval_ms :: pos_integer(), + timer_ref :: reference() | undefined +}). + +%% =================================================================== +%% API +%% =================================================================== + +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +%% @doc Run one cleanup pass (used by tests and manual ops). +-spec run_once() -> + #{notifications := non_neg_integer(), + auth_sessions := non_neg_integer(), + admin_audit := non_neg_integer()}. +run_once() -> + cleanup_worker(). + +%% @doc Returns a calendar:datetime() that is `Days` days before now. +-spec cutoff_datetime(non_neg_integer()) -> calendar:datetime(). +cutoff_datetime(Days) -> + Now = calendar:universal_time(), + Secs = calendar:datetime_to_gregorian_seconds(Now), + calendar:gregorian_seconds_to_datetime(Secs - Days * 86400). + +%% =================================================================== +%% gen_server callbacks +%% =================================================================== + +init([]) -> + process_flag(trap_exit, true), + IntervalMs = application:get_env(eventhub, cleanup_interval_ms, ?DEFAULT_INTERVAL_MS), + %% Schedule first run after a short delay to let the system stabilise + Ref = erlang:send_after(5000, self(), run_cleanup), + logger:info(#{what => infra_cleanup_started, interval_ms => IntervalMs}), + {ok, #state{interval_ms = IntervalMs, timer_ref = Ref}}. + +handle_call(run_once, _From, State) -> + {reply, cleanup_worker(), State}; +handle_call(_Msg, _From, State) -> + {reply, {error, unknown_call}, State}. + +handle_cast(_Msg, State) -> + {noreply, State}. + +handle_info(run_cleanup, #state{interval_ms = IntervalMs} = State) -> + spawn_link(fun cleanup_worker/0), + Ref = erlang:send_after(IntervalMs, self(), run_cleanup), + {noreply, State#state{timer_ref = Ref}}; + +handle_info({'EXIT', _Pid, normal}, State) -> + {noreply, State}; +handle_info({'EXIT', _Pid, Reason}, State) -> + logger:warning(#{what => infra_cleanup_worker_exit, reason => Reason}), + {noreply, State}; + +handle_info(_Msg, State) -> + {noreply, State}. + +terminate(_Reason, #state{timer_ref = Ref}) -> + case Ref of + undefined -> ok; + _ -> erlang:cancel_timer(Ref) + end, + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%% =================================================================== +%% Cleanup worker (runs in a spawned process or via run_once/0) +%% =================================================================== + +cleanup_worker() -> + CutoffNotif = cutoff_datetime(?NOTIFICATION_TTL_DAYS), + CutoffAuth = cutoff_datetime(?AUTH_SESSION_TTL_DAYS), + CutoffAudit = cutoff_datetime(?ADMIN_AUDIT_TTL_DAYS), + + N1 = cleanup_notifications(CutoffNotif), + N2 = cleanup_auth_sessions(CutoffAuth), + N3 = cleanup_admin_audit(CutoffAudit), + + logger:info(#{what => infra_cleanup_done, + notifications => N1, + auth_sessions => N2, + admin_audit => N3}), + #{notifications => N1, auth_sessions => N2, admin_audit => N3}. + +%% =================================================================== +%% Cleanup: notifications +%% =================================================================== + +cleanup_notifications(Cutoff) -> + Ms = ets:fun2ms(fun(#notification{id = Id, is_read = true, created_at = CreatedAt}) + when CreatedAt < Cutoff -> + Id + end), + delete_matching(notification, Ms). + +%% =================================================================== +%% Cleanup: auth_sessions +%% =================================================================== + +cleanup_auth_sessions(Cutoff) -> + MsRevoked = ets:fun2ms(fun(#auth_session{session_id = Id, + revoked = true, + created_at = CreatedAt}) + when CreatedAt < Cutoff -> + Id + end), + NRevoked = delete_matching(auth_session, MsRevoked), + Now = calendar:universal_time(), + MsExpired = ets:fun2ms(fun(#auth_session{session_id = Id, + expires_at = ExpiresAt, + created_at = CreatedAt}) + when CreatedAt < Cutoff, + ExpiresAt < Now -> + Id + end), + NExpired = delete_matching(auth_session, MsExpired), + NRevoked + NExpired. + +%% =================================================================== +%% Cleanup: admin_audit +%% =================================================================== + +cleanup_admin_audit(Cutoff) -> + Ms = ets:fun2ms(fun(#admin_audit{id = Id, timestamp = Ts}) + when Ts < Cutoff -> + Id + end), + delete_matching(admin_audit, Ms). + +%% =================================================================== +%% Chunked delete helpers +%% =================================================================== + +%% @doc Delete matching keys in separate transactions of ?CHUNK_SIZE, +%% up to ?MAX_CHUNKS_PER_TICK per call. Re-selects after each chunk +%% (continuation is not valid across transactions). +delete_matching(Table, Ms) -> + delete_matching_loop(Table, Ms, 0, ?MAX_CHUNKS_PER_TICK). + +delete_matching_loop(_Table, _Ms, Acc, 0) -> + Acc; +delete_matching_loop(Table, Ms, Acc, ChunksLeft) -> + case mnesia:transaction(fun() -> + case mnesia:select(Table, Ms, ?CHUNK_SIZE, write) of + '$end_of_table' -> + []; + {Ids, _Cont} -> + lists:foreach(fun(Id) -> mnesia:delete({Table, Id}) end, Ids), + Ids + end + end) of + {atomic, []} -> + Acc; + {atomic, Ids} when is_list(Ids) -> + delete_matching_loop(Table, Ms, Acc + length(Ids), ChunksLeft - 1); + {aborted, Reason} -> + logger:warning(#{what => infra_cleanup_tx_aborted, + table => Table, + reason => Reason}), + Acc + end. diff --git a/src/infra/infra_mnesia.erl b/src/infra/infra_mnesia.erl index e85e9b7..7933091 100755 --- a/src/infra/infra_mnesia.erl +++ b/src/infra/infra_mnesia.erl @@ -131,6 +131,7 @@ handle_call(init_tables, _From, State) -> end, ok = create_indices(), %% stats_collector:subscribe — после wait_for_tables + migrations (eventhub_app) + %% Table fragmentation: use infra_mnesia_fragmentation:fragment_table/2 manually. ok = start_cleanup_timer(), {reply, ok, State}; diff --git a/src/infra/infra_sup.erl b/src/infra/infra_sup.erl index 9a0eb42..b9f1dfa 100755 --- a/src/infra/infra_sup.erl +++ b/src/infra/infra_sup.erl @@ -56,6 +56,12 @@ init([]) -> restart => permanent, shutdown => 5000, type => worker, - modules => [subscription_worker]} + modules => [subscription_worker]}, + #{id => infra_cleanup, + start => {infra_cleanup, start_link, []}, + restart => permanent, + shutdown => 5000, + type => worker, + modules => [infra_cleanup]} ], - {ok, {SupFlags, Children}}. \ No newline at end of file + {ok, {SupFlags, Children}}. diff --git a/test/unit/infra_cleanup_tests.erl b/test/unit/infra_cleanup_tests.erl new file mode 100644 index 0000000..cf34223 --- /dev/null +++ b/test/unit/infra_cleanup_tests.erl @@ -0,0 +1,132 @@ +-module(infra_cleanup_tests). +-include_lib("eunit/include/eunit.hrl"). +-include("records.hrl"). + +-define(TABLES, [notification, auth_session, admin_audit]). + +setup() -> + eh_test_support:start_mnesia(), + eh_test_support:ensure_tables(?TABLES), + ok. + +cleanup(_) -> + eh_test_support:clear_tables(?TABLES), + eh_test_support:delete_tables(?TABLES), + eh_test_support:stop_mnesia(), + ok. + +infra_cleanup_test_() -> + {foreach, fun setup/0, fun cleanup/1, [ + {"cutoff_datetime is in the past", fun test_cutoff_datetime/0}, + {"keeps unread notifications", fun test_keeps_unread_notifications/0}, + {"deletes old read notifications", fun test_deletes_old_read_notifications/0}, + {"keeps active non-expired sessions", fun test_keeps_active_sessions/0}, + {"deletes old revoked sessions", fun test_deletes_old_revoked_sessions/0}, + {"deletes old admin_audit", fun test_deletes_old_admin_audit/0} + ]}. + +test_cutoff_datetime() -> + Cutoff = infra_cleanup:cutoff_datetime(30), + NowSecs = calendar:datetime_to_gregorian_seconds(calendar:universal_time()), + CutoffSecs = calendar:datetime_to_gregorian_seconds(Cutoff), + ?assert(CutoffSecs < NowSecs), + %% ~30 days (± a few seconds for test runtime) + DiffDays = (NowSecs - CutoffSecs) / 86400, + ?assert(DiffDays > 29.9), + ?assert(DiffDays < 30.1). + +test_keeps_unread_notifications() -> + Old = infra_cleanup:cutoff_datetime(60), + write_notification(<<"n_unread">>, false, Old), + write_notification(<<"n_read_recent">>, true, calendar:universal_time()), + #{notifications := N} = infra_cleanup:run_once(), + ?assertEqual(0, N), + ?assertMatch([_], mnesia:dirty_read(notification, <<"n_unread">>)), + ?assertMatch([_], mnesia:dirty_read(notification, <<"n_read_recent">>)). + +test_deletes_old_read_notifications() -> + Old = infra_cleanup:cutoff_datetime(60), + write_notification(<<"n_old_read">>, true, Old), + write_notification(<<"n_unread">>, false, Old), + #{notifications := N} = infra_cleanup:run_once(), + ?assertEqual(1, N), + ?assertEqual([], mnesia:dirty_read(notification, <<"n_old_read">>)), + ?assertMatch([_], mnesia:dirty_read(notification, <<"n_unread">>)). + +test_keeps_active_sessions() -> + Now = calendar:universal_time(), + Future = days_offset(Now, 7), + Recent = days_offset(Now, -1), + write_auth_session(<<"s_active">>, false, Future, Recent), + #{auth_sessions := N} = infra_cleanup:run_once(), + ?assertEqual(0, N), + ?assertMatch([_], mnesia:dirty_read(auth_session, <<"s_active">>)). + +test_deletes_old_revoked_sessions() -> + Now = calendar:universal_time(), + Old = days_offset(Now, -14), + Future = days_offset(Now, 7), + write_auth_session(<<"s_revoked_old">>, true, Future, Old), + write_auth_session(<<"s_active">>, false, Future, Old), + #{auth_sessions := N} = infra_cleanup:run_once(), + ?assert(N >= 1), + ?assertEqual([], mnesia:dirty_read(auth_session, <<"s_revoked_old">>)), + %% Active but old created_at with future expiry must stay + ?assertMatch([_], mnesia:dirty_read(auth_session, <<"s_active">>)). + +test_deletes_old_admin_audit() -> + Now = calendar:universal_time(), + Old = days_offset(Now, -120), + write_admin_audit(<<"a_old">>, Old), + write_admin_audit(<<"a_new">>, Now), + #{admin_audit := N} = infra_cleanup:run_once(), + ?assertEqual(1, N), + ?assertEqual([], mnesia:dirty_read(admin_audit, <<"a_old">>)), + ?assertMatch([_], mnesia:dirty_read(admin_audit, <<"a_new">>)). + +%%%=================================================================== +%%% Helpers +%%%=================================================================== + +write_notification(Id, IsRead, CreatedAt) -> + mnesia:dirty_write(#notification{ + id = Id, + user_id = <<"u1">>, + type = custom, + title = <<"t">>, + body = <<"b">>, + is_read = IsRead, + created_at = CreatedAt + }). + +write_auth_session(Id, Revoked, ExpiresAt, CreatedAt) -> + mnesia:dirty_write(#auth_session{ + session_id = Id, + family_id = <<"fam">>, + subject_id = <<"subj">>, + subject_type = user, + client_type = <<"web">>, + current_jti = <<"jti">>, + expires_at = ExpiresAt, + revoked = Revoked, + created_at = CreatedAt, + updated_at = CreatedAt + }). + +write_admin_audit(Id, Timestamp) -> + mnesia:dirty_write(#admin_audit{ + id = Id, + admin_id = <<"adm">>, + email = <<"a@test.local">>, + role = admin, + action = <<"test">>, + entity_type = <<"user">>, + entity_id = <<"u1">>, + timestamp = Timestamp, + ip = <<"127.0.0.1">>, + reason = <<>> + }). + +days_offset(Datetime, Days) -> + Secs = calendar:datetime_to_gregorian_seconds(Datetime), + calendar:gregorian_seconds_to_datetime(Secs + Days * 86400).