WebSocket (Jetstream / Firehose)

Overzicht

laravel-bluesky biedt twee WebSocket-commands om verbinding te maken met de realtime streams van Bluesky.

Jetstream — Het eigen, voorgefilterde WebSocket-endpoint van Bluesky. Lichtgewicht, in JSON-formaat.
Firehose — De ruwe eventstream van het AT Protocol. Ontvangt alle data in binair DAG-CBOR-formaat.

Voor langdurig draaiende processen via WebSocket heb je een permanent draaiende server zoals een VPS of EC2 of een custom worker op Laravel Cloud nodig. In serverless omgevingen zoals Laravel Vapor en Vercel werkt het niet.

Installatie

Voor de WebSocket-functionaliteit is Workerman nodig.

composer require workerman/workerman

Jetstream

Overzicht

Jetstream is de voorgefilterde WebSocket-service van Bluesky. Je kunt filteren op collectietype en gebruikers-DID, zodat je efficiënt alleen de events ontvangt die je nodig hebt.

Kenmerk	Beschrijving
Dataformaat	JSON
Filtering	Filterbaar op collectie en DID
Datavolume	Lichtgewicht, afhankelijk van de filters
Gebruik	Posts, likes, follows en dergelijke monitoren

Starten

# Alle berichten ontvangen (zonder filter)
php artisan bluesky:ws start

# Debuggen: alle ontvangen berichten weergeven
php artisan bluesky:ws start -v

Collectiefilter

Met de optie -C beperk je welke collecties je ontvangt. Je kunt er meerdere opgeven.

# Alleen posts en likes ontvangen
php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like

# Alleen follows ontvangen
php artisan bluesky:ws start -C app.bsky.graph.follow

Belangrijkste collecties:

Collectie	Inhoud
`app.bsky.feed.post`	Posts maken en verwijderen
`app.bsky.feed.like`	Likes
`app.bsky.feed.repost`	Reposts
`app.bsky.graph.follow`	Follows
`app.bsky.graph.block`	Blocks

DID-filter

Met de optie -D ontvang je alleen events van specifieke gebruikers.

# Alleen posts van specifieke gebruikers ontvangen
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

Events verwerken

Het Jetstream-command stuurt Laravel-events uit op basis van het type van het ontvangen bericht.

Eventklasse	Moment
`JetstreamMessageReceived`	Bij ontvangst van elk bericht
`JetstreamCommitMessage`	Bij het maken, bijwerken of verwijderen van een record
`JetstreamIdentityMessage`	Bij identity-events zoals een handlewijziging
`JetstreamAccountMessage`	Bij het activeren of deactiveren van een account

Maak een eventlistener om de events te verwerken.

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // Het collectietype controleren
        $collection = data_get($event->message, 'commit.collection');

        if ($collection !== 'app.bsky.feed.post') {
            return;
        }

        // Het type bewerking: create / update / delete
        $operation = $event->operation;

        // De DID van de auteur
        $did = $event->message['did'];

        // De inhoud van het record
        $record = data_get($event->message, 'commit.record');
        $text = data_get($record, 'text', '');

        info("[$operation] $did: $text");
    }
}

Firehose

Overzicht

De Firehose is de ruwe eventstream van het AT Protocol. Je ontvangt alle recordbewerkingen op het Bluesky-netwerk in binair (DAG-CBOR-)formaat.

Kenmerk	Beschrijving
Dataformaat	Binair DAG-CBOR (het pakket decodeert automatisch)
Filtering	Geen (je ontvangt alle data)
Datavolume	Zeer groot
Gebruik	Alle data verzamelen en archiveren

Het decoden van DAG-CBOR doet het pakket automatisch. In je eventlisteners ontvang je de data als gewone PHP-arrays.

Starten

php artisan bluesky:firehose start

# Debuggen: ontvangen berichten weergeven
php artisan bluesky:firehose start -v

Events verwerken

Ook het Firehose-command verwerkt berichten via Laravel-events.

Eventklasse	Moment
`FirehoseMessageReceived`	Bij ontvangst van elk bericht (inclusief ruwe data)
`FirehoseCommitMessage`	Bij het maken, bijwerken of verwijderen van een record
`FirehoseIdentityMessage`	Bij identity-events
`FirehoseAccountMessage`	Bij account-events
`FirehoseSyncMessage`	Bij repository-synchronisatie-events

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

class FirehosePostListener
{
    public function handle(FirehoseCommitMessage $event): void
    {
        // Het collectietype controleren
        if ($event->collection !== 'app.bsky.feed.post') {
            return;
        }

        // Het type bewerking: create / update / delete
        $action = $event->action;

        // De DID van de auteur
        $did = $event->did;

        // De inhoud van het record (gedecodeerde array)
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

        info("[$action] $did: $text");
    }
}

Configuratie

In config/bluesky.php kun je de host en de loginstellingen wijzigen.

// Jetstream
'jetstream' => [
    'host' => env('BLUESKY_JETSTREAM_HOST', 'jetstream1.us-west.bsky.network'),
    'max' => env('BLUESKY_JETSTREAM_MAX', 0), // maxMessageSizeBytes (0 = onbeperkt)
    'logging' => [
        'driver' => env('BLUESKY_JETSTREAM_LOG_DRIVER', 'daily'),
        'days' => 7,
        'path' => env('BLUESKY_JETSTREAM_LOG_PATH', storage_path('logs/jetstream.log')),
    ],
],

// Firehose
'firehose' => [
    'host' => env('BLUESKY_FIREHOSE_HOST', 'bsky.network'),
    'logging' => [
        'driver' => env('BLUESKY_FIREHOSE_LOG_DRIVER', 'daily'),
        'days' => 7,
        'path' => env('BLUESKY_FIREHOSE_LOG_PATH', storage_path('logs/firehose.log')),
    ],
],

Voorbeeldconfiguratie in .env:

BLUESKY_JETSTREAM_HOST=jetstream2.us-east.bsky.network
BLUESKY_JETSTREAM_MAX=1000000

Combineren met de Labeler

Je kunt de Labeler-server en Jetstream/Firehose tegelijk starten. De data van Jetstream of de Firehose kun je gebruiken bij het verwerken van labelverzoeken die de Labeler ontvangt.

# Labeler + Jetstream (follow-events monitoren)
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose (alle data ontvangen)
php artisan bluesky:labeler:server start --firehose

Zie de Labeler-pagina voor details over de Labeler.

Langdurig draaiende processen beheren

De WebSocket-commands zijn processen die lang blijven draaien. Gebruik in productie een procesmanager zoals Supervisor.

Voorbeeld van een Supervisor-configuratie

/etc/supervisor/conf.d/bluesky-jetstream.conf:

[program:bluesky-jetstream]
process_name=%(program_name)s_%(process_num)02d
command=php /var/www/html/artisan bluesky:ws start -C app.bsky.feed.post
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
user=www-data
numprocs=1
redirect_stderr=true
stdout_logfile=/var/www/html/storage/logs/jetstream-worker.log
stopwaitsecs=3600

# Supervisor opnieuw inlezen en starten
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

Daemon-configuratie in Laravel Forge

Gebruik je Laravel Forge, voeg dan een daemon toe via de sectie Daemons.

Command: php artisan bluesky:ws start -C app.bsky.feed.post
Directory: /var/www/html
User: forge

Achtergrondprocessen instellen op Laravel Cloud

De WebSocket-commands verbinden als WebSocket-client met de streams van Bluesky en werken daarom ook op Laravel Cloud. Stel ze in als achtergrondproces (custom worker) van Laravel Cloud. Voeg in de achtergrondprocesinstellingen van Laravel Cloud een Custom Worker toe. Voor Jetstream:

php artisan bluesky:ws start

Voor de Firehose:

php artisan bluesky:firehose start

Het stoppen en herstarten van processen bij een deploy wordt volledig automatisch afgehandeld door Laravel Cloud. Buiten het instellen van het achtergrondproces is er geen extra configuratie nodig.

Aandachtspunten

Als een proces onverwacht stopt, wordt het met autorestart=true automatisch herstart.
Overweeg periodieke herstarts om geheugenlekken te voorkomen.
Bij de Firehose, die enorme aantallen berichten ontvangt, wordt aanbevolen om de verwerking in je listener asynchroon te maken (met een queue-job).

// Voorbeeld: vanuit de listener dispatchen naar een queue-job

namespace App\Listeners;

use App\Jobs\ProcessFirehosePost;
use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

class FirehosePostListener
{
    public function handle(FirehoseCommitMessage $event): void
    {
        if ($event->collection !== 'app.bsky.feed.post') {
            return;
        }

        // Zware verwerking delegeren aan een queue-job
        ProcessFirehosePost::dispatch($event->did, $event->record, $event->action);
    }
}

Source: src/Console/WebSocket

GitHub Copilot

SDK

AI SDK

VOICEVOX

Starter kits

Feeds en RSS

Middleware

Overzicht

Installatie

Jetstream

Overzicht

Starten

Collectiefilter

DID-filter

Events verwerken

Firehose

Overzicht

Starten

Events verwerken

Configuratie

Combineren met de Labeler

Langdurig draaiende processen beheren

Voorbeeld van een Supervisor-configuratie

Daemon-configuratie in Laravel Forge

Achtergrondprocessen instellen op Laravel Cloud

Aandachtspunten

​Overzicht

​Installatie

​Jetstream

​Overzicht

​Starten

​Collectiefilter

​DID-filter

​Events verwerken

​Firehose

​Overzicht

​Starten

​Events verwerken

​Configuratie

​Combineren met de Labeler

​Langdurig draaiende processen beheren

​Voorbeeld van een Supervisor-configuratie

​Daemon-configuratie in Laravel Forge

​Achtergrondprocessen instellen op Laravel Cloud

​Aandachtspunten

Overzicht

Installatie

Jetstream

Overzicht

Starten

Collectiefilter

DID-filter

Events verwerken

Firehose

Overzicht

Starten

Events verwerken

Configuratie

Combineren met de Labeler

Langdurig draaiende processen beheren

Voorbeeld van een Supervisor-configuratie

Daemon-configuratie in Laravel Forge

Achtergrondprocessen instellen op Laravel Cloud

Aandachtspunten