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

@@ -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},

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.

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};
 

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}}.

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).