view release on metacpan or  search on metacpan

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

    return unless $data && %$data;

    # Deserialize rooms
    $data->{rooms} = $data->{rooms} ? $JSON->decode($data->{rooms}) : {};
    $data->{connected} = $data->{connected} ? 1 : 0;

    return $data;
}

async sub get_session_by_name {
    my ($name) = @_;

    # Scan for session with this name (not efficient, but works for demo)
    my $cursor = "0";
    do {
        my $result = await $redis->scan($cursor, MATCH => 'session:*', COUNT => 100);
        $cursor = $result->[0];
        my $keys = $result->[1] // [];

        for my $key (@$keys) {
            my $session = await $redis->hgetall($key);
            if ($session && $session->{name} eq $name && $session->{connected}) {
                $session->{rooms} = $session->{rooms} ? $JSON->decode($session->{rooms}) : {};
                return $session;
            }
        }
    } while ($cursor && $cursor ne "0");

    return;
}

async sub create_session {
    my ($session_id, $name, $send_cb) = @_;

    my $session = {
        id        => $session_id,
        name      => $name,
        connected => 1,
        joined_at => time(),
        last_seen => time(),
        rooms     => {},
    };

    await $redis->hmset("session:$session_id",
        id        => $session_id,
        name      => $name,
        connected => 1,
        joined_at => $session->{joined_at},
        last_seen => $session->{last_seen},
        rooms     => '{}',
    );
    await $redis->expire("session:$session_id", SESSION_TTL);
    await $redis->sadd('connected:sessions', $session_id);

    # Track locally for this worker
    register_local_session($session_id, $send_cb);

    return $session;
}

async sub update_session {
    my ($session_id, $updates) = @_;

    my @args;
    for my $key (keys %$updates) {
        my $val = $updates->{$key};
        $val = $JSON->encode($val) if ref $val;
        push @args, $key, $val;
    }

    await $redis->hmset("session:$session_id", @args) if @args;
    await $redis->expire("session:$session_id", SESSION_TTL);
}

async sub set_session_connected {
    my ($session_id, $send_cb) = @_;

    await update_session($session_id, { connected => 1, last_seen => time() });
    await $redis->sadd('connected:sessions', $session_id);
    register_local_session($session_id, $send_cb);

    return await get_session($session_id);
}

async sub set_session_disconnected {
    my ($session_id) = @_;

    await update_session($session_id, { connected => 0 });
    await $redis->srem('connected:sessions', $session_id);
    unregister_local_session($session_id);
}

async sub is_session_connected {
    my ($session_id) = @_;
    my $connected = await $redis->hget("session:$session_id", 'connected');
    return $connected ? 1 : 0;
}

async sub remove_session {
    my ($session_id) = @_;

    my $session = await get_session($session_id);
    return unless $session;

    # Leave all rooms
    for my $room_name (keys %{$session->{rooms}}) {
        await remove_user_from_room($session_id, $room_name, 1);
    }

    await $redis->del("session:$session_id");
    await $redis->srem('connected:sessions', $session_id);
    unregister_local_session($session_id);

    return $session;
}

# Room management
async sub get_room {
    my ($name) = @_;

    my $exists = await $redis->exists("room:$name");