WebSocket using PAGI::FastAPI

$ pagi-server websocket_app.pl

$ docker compose up -d

my $app = PAGI::FastAPI->new(
    title   => 'PAGI::FastAPI Chat Server',
    version => '1.0.0',
);

$app->on_startup(async sub {
    print "Server starting up...\n";

    await pg_query(q{
        CREATE TABLE IF NOT EXISTS chat_users (
            session_id TEXT PRIMARY KEY,
            username   TEXT NOT NULL,
            last_seen  TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    });

    # Stale-user cleanup, every 60s. Same Future::IO approach as the
    # heartbeat above, started here, stopped via the $shutting_down flag
    # in on_shutdown rather than an explicit timer object.
    cleanup_loop()->retain;

    # Cross-process fan-out: other server processes publish here too, so a
    # message sent to the server on :3000 reaches clients connected to :3001.
    $pubsub->listen('chat_messages' => sub {
        my ($pubsub, $payload) = @_;

        my $data = ref($payload) eq 'HASH'
            ? $payload
            : eval { decode_json($payload) };
        return unless $data;

        my $from_process = delete $data->{_process_id};
        return if defined $from_process && $from_process eq $process_id;

        if ($data->{type} eq 'message') {
            push @history, $data;
            shift @history if @history > 10;
        }

        foreach my $client (values %$clients) {
            eval { $client->{ws}->send_json($data) };
        }
    });
});

$app->on_shutdown(async sub {
    print "Server shutting down...\n";

    $shutting_down = 1;

    # Instantly wake every heartbeat/cleanup coroutine currently sleeping,
    # instead of leaving them to notice $shutting_down up to 30-60s later.
    for my $sleep (values %active_sleeps) {
        $sleep->cancel unless $sleep->is_ready;
    }

    # Cancelling sleeps stops *new*  queries from starting, but doesn't
    # touch a query already sent to Postgres, wait (up to 5s) for those
    # to actually get a response and  resolve their Future normally, so
    # we don't disconnect out from under one and risk the same class of
    # use-after-free during global destruction.
    await wait_for_inflight_queries();

    eval { $pubsub->unlisten('chat_messages') };
    eval { $pg->db->disconnect };
});

$app->websocket('/chat',
    handler => async sub ($ws, $deps) {
        await $ws->accept;

        my $id = "$ws";
        $clients->{$id} = { ws => $ws, name => 'Anonymous' };

        # Heartbeat: keeps this connection's last_seen fresh in Postgres
        # for cross-process presence. Future::IO->sleep is loop-agnostic,
        # no IO::Async::Loop object of our own is needed, since it delegates
        # to whatever Future::IO backend is already configured (PAGI::Server
        # provides one). The loop naturally stops once $clients->{$id} is
        # deleted on disconnect below, no explicit timer teardown needed.
        heartbeat_loop($id)->else(sub { Future->done })->retain;

        # Handle incoming messages
        while (1) {
            my $msg_text = await $ws->receive_text;
            last unless defined $msg_text;

            my $data = eval { decode_json($msg_text) };
            next unless $data;

            if ($data->{type} eq 'typing') {
                await broadcast({
                    type     => 'typing',
                    user     => $clients->{$id}{name},
                    isTyping => $data->{isTyping} ? 1 : 0
                }, $id);
            }
            elsif ($data->{type} eq 'join') {
                $clients->{$id}{name} = $data->{name};

                await pg_query(q{
                    INSERT INTO chat_users (session_id, username, last_seen)
                    VALUES (?, ?, NOW())
                    ON CONFLICT (session_id)
                    DO UPDATE SET username  = EXCLUDED.username,
                                  last_seen = NOW()
                }, $id, $data->{name});

                # Send history to new user
                foreach my $old_msg (@history) {
                    await $ws->send_text(encode_json($old_msg));
                }

                await broadcast({
                    type => 'system',
                    text => "$data->{name} joined"
                });
                await send_user_list();
            }
            elsif ($data->{type} eq 'message') {
                my (undef, $min, $hour) = localtime();
                my $timestamp = sprintf("%02d:%02d", $hour, $min);
                my $msg_out = {
                    type      => 'message',
                    user      => $clients->{$id}{name},
                    text      => $data->{text},
                    timestamp => $timestamp
                };

                push @history, $msg_out;
                shift @history if @history > 10;

                await broadcast($msg_out);
            }
        }

        # Cleanup on disconnect
        my $name = $clients->{$id}{name};
        delete $clients->{$id};
        await pg_query(q{DELETE FROM chat_users WHERE session_id = ?}, $id);
        await broadcast({ type => 'system', text => "$name left" });
        await send_user_list();
    }
);

$ IO_ASYNC_LOOP=EV pagi-server chat-server-v6.pl
Future::IO configured for IO::Async
PAGI development mode - Lint middleware enabled
Server starting up...
access_log is a terminal; this may impact performance. Consider redirecting to a file or setting access_log => undef for benchmarks.
PAGI Server listening on http://127.0.0.1:5000/ (loop: EV, max_conn: 1000, http2: not installed, tls: not installed, future_xs: available)