firehose

#!/usr/bin/env raku

use Cro::WebSocket::Client;
use NCurses;

my $win = initscr;
die "Failed to initialize ncurses\n" unless $win.defined;


my $client = Cro::WebSocket::Client.new: :json, uri => 'wss://jetstream1.us-east.bsky.network/subscribe';

my $conn = await $client.connect;

my $lock = Lock.new;
my %counts;
my $counts-supplier = Supplier.new;
react {
	whenever $conn.messages -> $message {
        whenever $message.body -> $body {
            $lock.protect: {
                if $body<commit><collection>:exists {
                    %counts{$body<commit><collection>}++;
                }
            }
        }
	}
    whenever Supply.interval(1) {
        $lock.protect: {
            $counts-supplier.emit: %counts.clone;
            %counts = ();
        }
    }
    whenever $counts-supplier.Supply -> %count {
        clear();
        mvaddstr( 3, 5, "Post:   { %count<app.bsky.feed.post> // 0 }");
        mvaddstr( 4, 5, "Like:   { %count<app.bsky.feed.like>  //0 }");
        mvaddstr( 5, 5, "Repost: { %count<app.bsky.feed.repost> // 0 }");
        mvaddstr( 6, 5, "Block:  { %count<app.bsky.graph.block> // 0 }");
        mvaddstr( 7, 5, "Follow: { %count<app.bsky.graph.follow> // 0 }");
        nc_refresh;
    }
    whenever signal(SIGINT) {
        endwin();
        done;
    }
}