Ü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
| 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 |
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')),
],
],
.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
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
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);
}
}
Zuletzt geändert am 13. Juli 2026
