view release on metacpan or  search on metacpan

examples/pagi-chat/lib/ChatApp/State.pm  view on Meta::CPAN

our @EXPORT_OK = qw(
    init_redis get_redis get_pubsub
    get_session create_session update_session remove_session
    get_session_by_name set_session_connected set_session_disconnected
    is_session_connected
    get_room add_room remove_room get_all_rooms
    add_user_to_room remove_user_from_room get_room_users
    add_message get_room_messages
    get_stats generate_id sanitize_username sanitize_room_name
    subscribe_broadcasts register_local_session unregister_local_session
    broadcast_to_room broadcast_global add_local_room
    add_background_task
);

my $JSON = JSON::MaybeXS->new->utf8->canonical;

# Redis connections (per-worker)
my $redis;
my $pubsub;
my $pubsub_subscription;

# Background task selector (per-worker)
my $background_selector;

# Local session callbacks (for this worker only)
# Redis PubSub delivers to all workers, but we only call callbacks for OUR clients
my %local_sessions;

use constant {
    MAX_MESSAGES_PER_ROOM => 100,
    SESSION_TTL           => 86400,    # 24 hours
    BROADCAST_CHANNEL     => 'chat:broadcast',
};

# Track server start time for uptime
my $server_start_time = time();

# Initialize Redis connections for this worker
async sub init_redis {
    my (%args) = @_;

    require Async::Redis;

    my $host = $args{host} // $ENV{REDIS_HOST} // 'localhost';
    my $port = $args{port} // $ENV{REDIS_PORT} // 6379;

    # Main connection for commands
    $redis = Async::Redis->new(host => $host, port => $port);
    await $redis->connect;

    # Separate connection for PubSub
    my $pubsub_redis = Async::Redis->new(host => $host, port => $port);
    await $pubsub_redis->connect;
    $pubsub = await $pubsub_redis->subscribe(BROADCAST_CHANNEL);

    # Initialize background task selector and start the runner
    $background_selector = Future::Selector->new;
    _start_selector_runner();

    # Start periodic stats timer (every 10 seconds)
    _start_stats_timer();

    # Initialize default rooms
    await add_room('general', 'system');
    await add_room('random', 'system');
    await add_room('help', 'system');

    print STDERR "Worker $$: Redis state initialized\n";
    return $redis;
}

sub get_redis { $redis }
sub get_pubsub { $pubsub }

# Background selector runner
my $selector_runner_future;

# Periodic stats timer
my $stats_timer;

# Start the selector with the PubSub listener as the main long-running task
sub _start_selector_runner {
    # Add the broadcast listener as a long-running task
    # Use gen => to provide a generator that creates the future
    # (f => expects a completed future, gen => expects a coderef)
    $background_selector->add(
        data => 'pubsub-listener',
        gen  => sub { _broadcast_listener() },
    );

    # Run the selector in the background
    $selector_runner_future = $background_selector->run->on_fail(sub {
        my ($err) = @_;
        warn "[selector] Runner failed: $err";
    })->retain;
}

# Start periodic stats timer using Future::IO (event-loop agnostic)
sub _start_stats_timer {
    $stats_timer = (async sub {
        while (1) {
            await Future::IO->sleep(10);
            next unless %local_sessions;  # Skip if no clients
            eval {
                my $stats = await get_stats();
                my $msg = $JSON->encode({
                    global  => 1,
                    payload => {
                        type         => 'stats',
                        users_online => $stats->{users_online},
                        rooms_count  => $stats->{rooms_count},
                        uptime       => $stats->{uptime},
                    },
                });
                await $redis->publish(BROADCAST_CHANNEL, $msg);
            };
            warn "[stats] Timer error: $@" if $@;
        }
    })->()->retain;
}

# Add a fire-and-forget background task to the selector
# Pass a coderef that returns a future, not the future itself
sub add_background_task {
    my ($gen, $description) = @_;
    $description //= 'background task';

    return unless $background_selector;

    # gen => expects a coderef that generates futures
    # It will be called once, and again if the future completes (until it returns undef)
    my $called = 0;
    $background_selector->add(
        data => $description,
        gen  => sub {
            return undef if $called++;  # One-shot: only generate once
            return ref($gen) eq 'CODE' ? $gen->() : $gen;
        },
    );
}

# Long-running broadcast listener (async sub with while loop)
async sub _broadcast_listener {
    print STDERR "[pubsub] Worker $$: Broadcast listener started\n";

    while (my $msg = await $pubsub->next_message) {
        next unless $msg->{type} eq 'message';

        my $data = eval { $JSON->decode($msg->{message}) };
        next unless $data;

        my $payload = $data->{payload};

        # Global broadcast - deliver to ALL local sessions
        if ($data->{global}) {
            for my $session_id (keys %local_sessions) {
                my $local = $local_sessions{$session_id};
                next unless $local && $local->{send_cb};

                eval { $local->{send_cb}->($payload) };