WebSocket (Jetstream / Firehose)

Fonctionnalités WebSocket de Laravel Bluesky. Traitement des événements en temps réel avec Bluesky Jetstream et AT Protocol Firehose. Utilisation de JetstreamServeCommand et FirehoseServeCommand.

​

Vue d’ensemble

laravel-bluesky fournit deux commandes WebSocket permettant de se connecter aux flux en temps réel de Bluesky.

• Jetstream — Endpoint WebSocket propre à Bluesky avec filtrage préappliqué. Léger, au format JSON.
• Firehose — Flux d’événements bruts d’AT Protocol. Reçoit toutes les données au format binaire DAG-CBOR.

Les processus WebSocket de longue durée nécessitent un serveur toujours actif comme un VPS ou EC2 ou un worker personnalisé Laravel Cloud. Ils ne fonctionnent pas dans des environnements serverless comme Laravel Vapor ou Vercel.

​

Installation

La fonctionnalité WebSocket requiert Workerman.

composer require workerman/workerman

​

Jetstream

​

Vue d’ensemble

Jetstream est un service WebSocket avec filtrage préappliqué fourni par Bluesky. Vous pouvez filtrer par type de collection ou par DID utilisateur, ce qui permet de recevoir efficacement uniquement les événements nécessaires.

Caractéristique	Description
Format des données	JSON
Filtrage	Filtrable par collection et DID
Volume de données	Léger selon le filtre
Utilisation	Surveillance des publications, likes, follows, etc.

​

Méthode de démarrage

# Recevoir tous les messages (sans filtre)
php artisan bluesky:ws start

# Débogage : afficher tous les messages reçus
php artisan bluesky:ws start -v

​

Filtre par collection

L’option -C restreint les collections à recevoir. Plusieurs valeurs sont possibles.

# Recevoir uniquement les publications et likes
php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like

# Recevoir uniquement les follows
php artisan bluesky:ws start -C app.bsky.graph.follow

Principales collections :

Collection	Contenu
app.bsky.feed.post	Création/suppression de publications
app.bsky.feed.like	Likes
app.bsky.feed.repost	Reposts
app.bsky.graph.follow	Follows
app.bsky.graph.block	Blocages

​

Filtre par DID

L’option -D permet de recevoir uniquement les événements d’utilisateurs spécifiques.

# Recevoir uniquement les publications d'utilisateurs spécifiques
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

​

Traitement des événements

La commande Jetstream déclenche des événements Laravel selon le type de message reçu.

Classe d’événement	Déclenchement
JetstreamMessageReceived	À la réception de tout message
JetstreamCommitMessage	Création, mise à jour ou suppression d’un enregistrement
JetstreamIdentityMessage	Événements d’identité comme un changement de handle
JetstreamAccountMessage	Activation/désactivation de compte

Créez un listener d’événement pour traiter les événements.

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // Vérifier le type de collection
        $collection = data_get($event->message, 'commit.collection');

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

        // Type d'opération : create / update / delete
        $operation = $event->operation;

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

        // Contenu de l'enregistrement
        $record = data_get($event->message, 'commit.record');
        $text = data_get($record, 'text', '');

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

​

Firehose

​

Vue d’ensemble

Firehose est le flux brut d’événements d’AT Protocol. Il reçoit toutes les opérations d’enregistrements du réseau Bluesky au format binaire (DAG-CBOR).

Caractéristique	Description
Format des données	Binaire DAG-CBOR (le package décode automatiquement)
Filtrage	Aucun (reçoit toutes les données)
Volume de données	Très important
Utilisation	Collecte et archivage de toutes les données

Le décodage DAG-CBOR est effectué automatiquement par le package. Dans les listeners d’événements, vous recevez les données sous forme de tableau PHP standard.

​

Méthode de démarrage

php artisan bluesky:firehose start

# Débogage : afficher les messages reçus
php artisan bluesky:firehose start -v

​

Traitement des événements

La commande Firehose utilise également les événements Laravel pour traiter les messages.

Classe d’événement	Déclenchement
FirehoseMessageReceived	À la réception de tout message (incluant les données brutes)
FirehoseCommitMessage	Création, mise à jour ou suppression d’un enregistrement
FirehoseIdentityMessage	Événements d’identité
FirehoseAccountMessage	Événements de compte
FirehoseSyncMessage	Événements de synchronisation du dépôt

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

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

        // Type d'opération : create / update / delete
        $action = $event->action;

        // DID de l'auteur
        $did = $event->did;

        // Contenu de l'enregistrement (tableau décodé)
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

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

​

Configuration

Vous pouvez modifier l’hôte de connexion et la configuration des logs dans config/bluesky.php.

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

Exemple de configuration dans .env :

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

​

Combinaison avec Labeler

Vous pouvez démarrer simultanément un serveur Labeler et Jetstream / Firehose. Les données de Jetstream ou Firehose peuvent être utilisées pour traiter les requêtes d’étiquetage reçues par le Labeler.

# Labeler + Jetstream (surveillance des événements follow)
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose (recevoir toutes les données)
php artisan bluesky:labeler:server start --firehose

Pour plus de détails sur Labeler, consultez la page Labeler.

​

Exploitation des processus de longue durée

Les commandes WebSocket sont des processus qui restent actifs longtemps. En production, utilisez un outil de gestion de processus comme Supervisor.

​

Exemple de configuration Supervisor

/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

# Recharger et démarrer Supervisor
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

​

Configuration du daemon Laravel Forge

Si vous utilisez Laravel Forge, ajoutez un daemon depuis la section Daemons.

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

​

Configuration des processus en arrière-plan sur Laravel Cloud

Les commandes WebSocket se connectent au flux Bluesky en tant que client WebSocket, elles fonctionnent donc aussi sur Laravel Cloud. Configurez-les comme processus en arrière-plan (worker personnalisé) de Laravel Cloud.

Ajoutez un Custom Worker dans la configuration des processus en arrière-plan de Laravel Cloud.

Pour Jetstream :

php artisan bluesky:ws start

Pour Firehose :

php artisan bluesky:firehose start

Les arrêts et redémarrages de processus lors des déploiements sont tous gérés automatiquement par Laravel Cloud. Aucune configuration supplémentaire n’est nécessaire en dehors de la configuration du processus en arrière-plan.

​

Points d’attention

• Si le processus se termine de manière inattendue, il est redémarré automatiquement grâce à autorestart=true.
• Pour éviter les fuites mémoire, envisagez des redémarrages périodiques.
• Pour Firehose, qui reçoit un grand nombre de messages, il est recommandé de rendre asynchrone le traitement dans les listeners (Queue Job).

// Exemple : dispatcher vers un Queue Job depuis le listener

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

        // Déléguer le traitement lourd à un Queue Job
        ProcessFirehosePost::dispatch($event->did, $event->record, $event->action);
    }
}

Source: src/Console/WebSocket