%% Bounded subscriber queues. Publishers admit directly into ETS; subscriber
%% mailboxes carry only a coalesced wake, never event payloads.
-module(albedo_bus).
-export([subscribe/2, drain/1, rearm/2, publish/1, has_subscribers/0,
         mark_running/2, running/1, forget/1]).

-define(TABLE, albedo_bus).
-define(QUEUES, albedo_bus_queues).
-define(EVENT_LIMIT, 256).
-define(BYTE_LIMIT, 1048576).
-define(CAS_ATTEMPTS, 32).

subscribe(Owner, Notify) ->
    Ref = make_ref(),
    Control = atomics:new(2, [{signed, false}]),
    albedo_registry:register(?QUEUES, Ref, {queue:new(), 0, 0}),
    albedo_registry:register(?TABLE, Ref, {Owner, Notify, Control}),
    spawn(fun() ->
        Monitor = erlang:monitor(process, Owner),
        receive {'DOWN', Monitor, process, Owner, _} ->
            ets:delete(?TABLE, Ref),
            ets:delete(?QUEUES, Ref)
        end
    end),
    Ref.

has_subscribers() ->
    case ets:info(?TABLE, size) of
        Size when is_integer(Size), Size > 0 -> true;
        _ -> false
    end.

publish(Event) ->
    case ets:whereis(?TABLE) of
        undefined -> nil;
        _ ->
            lists:foreach(
                fun({Ref, {_Owner, Notify, Control}}) ->
                    admit(Ref, Event, byte_size(Event), Notify, Control, ?CAS_ATTEMPTS)
                end,
                ets:tab2list(?TABLE)),
            nil
    end.

admit(Ref, Event, Size, Notify, Control, Attempts) ->
    case atomics:get(Control, 1) of
        1 -> ok;
        0 when Attempts == 0 -> overflow(Ref, Notify, Control);
        0 ->
            case ets:lookup(?QUEUES, Ref) of
                [] -> ok;
                [{Ref, {Queue, Count, Bytes} = Previous}] ->
                    case Count + 1 > ?EVENT_LIMIT orelse Bytes + Size > ?BYTE_LIMIT of
                        true ->
                            case replace_queue(Ref, Previous, Previous) of
                                true -> overflow(Ref, Notify, Control);
                                false -> admit(Ref, Event, Size, Notify, Control, Attempts - 1)
                            end;
                        false ->
                            Next = {queue:in(Event, Queue), Count + 1, Bytes + Size},
                            case replace_queue(Ref, Previous, Next) of
                                true -> wake(Notify, Control);
                                false -> admit(Ref, Event, Size, Notify, Control, Attempts - 1)
                            end
                    end
            end
    end.

%% The separate atomic latch makes overload finite even under competing
%% publishers. A stale CAS can only replace an existing row, so deletion
%% prevents an in-progress admission from recreating an overflowed queue.
overflow(Ref, Notify, Control) ->
    case atomics:compare_exchange(Control, 1, 0, 1) of
        ok -> ets:delete(?QUEUES, Ref), wake(Notify, Control);
        _ -> ok
    end.

wake(Notify, Control) ->
    case atomics:compare_exchange(Control, 2, 0, 1) of
        ok -> try Notify() catch _:_ -> ok end;
        _ -> ok
    end.

replace_queue(Ref, Previous, Next) ->
    ets:select_replace(?QUEUES, [
        {{Ref, '$1'}, [{'=:=', '$1', {const, Previous}}], [{const, {Ref, Next}}]}
    ]) == 1.

drain(Ref) ->
    case ets:lookup(?TABLE, Ref) of
        [{Ref, {_Owner, Notify, Control}}] -> drain(Ref, Notify, Control, ?CAS_ATTEMPTS);
        [] -> {batch, []}
    end.

drain(Ref, Notify, Control, Attempts) ->
    case atomics:get(Control, 1) of
        1 -> ets:delete(?QUEUES, Ref), overflow;
        0 when Attempts == 0 -> overflow(Ref, Notify, Control), overflow;
        0 ->
            case ets:lookup(?QUEUES, Ref) of
                [] -> {batch, []};
                [{Ref, {Queue, _Count, _Bytes} = Previous}] ->
                    case replace_queue(Ref, Previous, {queue:new(), 0, 0}) of
                        false -> drain(Ref, Notify, Control, Attempts - 1);
                        true ->
                            case atomics:get(Control, 1) of
                                1 -> ets:delete(?QUEUES, Ref), overflow;
                                0 -> {batch, queue:to_list(Queue)}
                            end
                    end
            end
    end.

%% A periodic flush can overtake the wake already in the stream mailbox.
%% Only consuming that wake permits rearming; otherwise its gate stays set.
rearm(Ref, WakeConsumed) ->
    case ets:lookup(?TABLE, Ref) of
        [{Ref, {_Owner, Notify, Control}}] ->
            case WakeConsumed of
                true -> atomics:put(Control, 2, 0);
                false -> ok
            end,
            case {atomics:get(Control, 1), ets:lookup(?QUEUES, Ref)} of
                {1, _} -> wake(Notify, Control);
                {0, [{Ref, {_Queue, Count, _Bytes}}]} when Count > 0 -> wake(Notify, Control);
                _ -> ok
            end;
        [] -> ok
    end,
    nil.

%% Whether each session is running a turn, as its actor last announced. The
%% registry reads this instead of asking a session, which may be busy.
mark_running(Session, Running) ->
    albedo_registry:register(albedo_running, Session, Running).

running(Session) ->
    case albedo_registry:lookup(albedo_running, Session) of
        {ok, true} -> true;
        _ -> false
    end.

forget(Session) -> albedo_registry:forget(albedo_running, Session).