WebSocket (Jetstream / Firehose)

Die WebSocket-Funktionen von Laravel Bluesky. Echtzeit-Event-Verarbeitung über Bluesky Jetstream und AT-Protocol-Firehose. Verwendung der Kommandos JetstreamServeCommand und FirehoseServeCommand.

​

Überblick

laravel-bluesky stellt zwei WebSocket-Kommandos zur Verbindung mit Blueskys Echtzeit-Stream bereit.

• Jetstream – Blueskys eigener, vorgefilterter WebSocket-Endpunkt. Leichtgewichtig im JSON-Format.
• Firehose – Der rohe Eventstream des AT Protocols. Empfängt alle Daten im binären DAG-CBOR-Format.

Für langlaufende Prozesse per WebSocket benötigen Sie einen dauerhaft laufenden Server (VPS, EC2 o. Ä.) oder einen Custom Worker in Laravel Cloud. In serverlosen Umgebungen wie Laravel Vapor oder Vercel funktioniert das nicht.

​

Installation

Für die WebSocket-Funktionen wird Workerman benötigt.

composer require workerman/workerman

​

Jetstream

​

Überblick

Jetstream ist ein von Bluesky bereitgestellter, vorgefilterter WebSocket-Dienst. Sie können nach Collection-Typ oder Nutzer-DID filtern und so gezielt nur die benötigten Events empfangen.

Merkmal	Beschreibung
Datenformat	JSON
Filterung	Filter nach Collection und DID möglich
Datenvolumen	Je nach Filter leichtgewichtig
Anwendung	Beobachtung von Posts, Likes, Follows usw.

​

Start

# Alle Nachrichten empfangen (ohne Filter)
php artisan bluesky:ws start

# Debug: Alle empfangenen Nachrichten anzeigen
php artisan bluesky:ws start -v

​

Collection-Filter

Mit der Option -C beschränken Sie die empfangenen Collections. Mehrfachangabe möglich.

# Nur Posts und Likes empfangen
php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like

# Nur Follows empfangen
php artisan bluesky:ws start -C app.bsky.graph.follow

Wichtige Collections:

Collection	Inhalt
app.bsky.feed.post	Erstellen/Löschen von Posts
app.bsky.feed.like	Likes
app.bsky.feed.repost	Reposts
app.bsky.graph.follow	Follows
app.bsky.graph.block	Blocks

​

DID-Filter

Mit der Option -D empfangen Sie nur Events bestimmter Nutzer.

# Nur Posts bestimmter Nutzer empfangen
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

​

Event-Verarbeitung

Das Jetstream-Kommando löst je nach empfangener Nachricht Laravel-Events aus.

Event-Klasse	Zeitpunkt
JetstreamMessageReceived	Beim Empfang jeder Nachricht
JetstreamCommitMessage	Beim Erstellen, Aktualisieren oder Löschen von Records
JetstreamIdentityMessage	Bei Identity-Events wie Handle-Änderungen
JetstreamAccountMessage	Beim Aktivieren/Deaktivieren eines Accounts

Erstellen Sie einen Event-Listener zur Verarbeitung.

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // Collection-Typ prüfen
        $collection = data_get($event->message, 'commit.collection');

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

        // Aktionstyp: create / update / delete
        $operation = $event->operation;

        // DID des Autors
        $did = $event->message['did'];

        // Inhalt des Records
        $record = data_get($event->message, 'commit.record');
        $text = data_get($record, 'text', '');

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

​

Firehose

​

Überblick

Firehose ist der rohe Eventstream des AT Protocols. Er empfängt sämtliche Record-Operationen im Bluesky-Netzwerk in binärer Form (DAG-CBOR).

Merkmal	Beschreibung
Datenformat	Binäres DAG-CBOR (automatische Decodierung durch das Paket)
Filterung	Nein (alle Daten werden empfangen)
Datenvolumen	Sehr groß
Anwendung	Sammeln/Archivieren aller Daten

Das Decodieren von DAG-CBOR erledigt das Paket automatisch. In Event-Listenern erhalten Sie die Daten als gewöhnliches PHP-Array.

​

Start

php artisan bluesky:firehose start

# Debug: Empfangene Nachrichten anzeigen
php artisan bluesky:firehose start -v

​

Event-Verarbeitung

Auch das Firehose-Kommando verarbeitet Nachrichten über Laravel-Events.

Event-Klasse	Zeitpunkt
FirehoseMessageReceived	Beim Empfang jeder Nachricht (inkl. Rohdaten)
FirehoseCommitMessage	Beim Erstellen, Aktualisieren oder Löschen von Records
FirehoseIdentityMessage	Bei Identity-Events
FirehoseAccountMessage	Bei Account-Events
FirehoseSyncMessage	Bei Repository-Sync-Events

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

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

        // Aktionstyp: create / update / delete
        $action = $event->action;

        // DID des Autors
        $did = $event->did;

        // Inhalt des Records (bereits decodiertes Array)
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

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

​

Konfiguration

In config/bluesky.php können Sie Zielhosts und Log-Einstellungen anpassen.

// Jetstream
'jetstream' => [
    'host' => env('BLUESKY_JETSTREAM_HOST', 'jetstream1.us-west.bsky.network'),
    'max' => env('BLUESKY_JETSTREAM_MAX', 0), // maxMessageSizeBytes (0 = unbegrenzt)
    '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')),
    ],
],

Beispiel für .env:

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

​

Kombination mit Labeler

Labeler-Server und Jetstream/Firehose lassen sich gemeinsam starten. Daten aus Jetstream oder Firehose können für die Verarbeitung von Labeling-Requests im Labeler genutzt werden.

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

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

Details zum Labeler finden Sie auf der Labeler-Seite.

​

Betrieb langlaufender Prozesse

WebSocket-Kommandos sind dauerhaft laufende Prozesse. Verwenden Sie in Produktivumgebungen ein Prozess-Management-Tool wie Supervisor.

​

Supervisor-Konfigurationsbeispiel

/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 neu einlesen und starten
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

​

Daemon-Konfiguration in Laravel Forge

Wenn Sie Laravel Forge verwenden, fügen Sie den Daemon im Bereich Daemons hinzu.

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

​

Konfiguration als Hintergrundprozess in Laravel Cloud

Die WebSocket-Kommandos verbinden sich als WebSocket-Clients mit den Bluesky-Streams und funktionieren daher auch in Laravel Cloud. Konfigurieren Sie sie als Hintergrundprozess (Custom Worker) in Laravel Cloud.

Fügen Sie in der Hintergrundprozess-Konfiguration von Laravel Cloud einen Custom Worker hinzu.

Für Jetstream:

php artisan bluesky:ws start

Für Firehose:

php artisan bluesky:firehose start

Prozessstopps und Neustarts beim Deployment übernimmt Laravel Cloud automatisch. Außer der Hintergrundprozess-Konfiguration ist keine weitere Einrichtung nötig.

​

Hinweise

• Beendet sich der Prozess unerwartet, wird er über autorestart=true automatisch neu gestartet.
• Um Memory-Leaks vorzubeugen, sollten Sie regelmäßige Neustarts in Erwägung ziehen.
• Bei Firehose mit hohem Nachrichtenaufkommen empfiehlt sich, die Verarbeitung im Listener asynchron (über einen Queue-Job) zu erledigen.

// Beispiel: Aus dem Listener einen Queue-Job dispatchen

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;
        }

        // Aufwendige Verarbeitung an Queue-Job delegieren
        ProcessFirehosePost::dispatch($event->did, $event->record, $event->action);
    }
}

Source: src/Console/WebSocket