WebSocket (Jetstream / Firehose) - Knowledge of Laravel

概览

laravel-bluesky 提供了两条 WebSocket 命令，用来连接 Bluesky 的实时事件流。

Jetstream —— Bluesky 提供的、已经过滤的 WebSocket 端点。JSON 格式，比较轻量。
Firehose —— AT Protocol 的原始事件流。以 DAG-CBOR 二进制格式接收全部数据。

基于 WebSocket 的长时运行进程需要 VPS 或 EC2 等常驻服务器，或者 Laravel Cloud 的自定义 worker。它不能在 Laravel Vapor、Vercel 等无服务器环境中运行。

安装

WebSocket 功能依赖 Workerman。

composer require workerman/workerman

Jetstream

概览

Jetstream 是 Bluesky 提供的、已经预先过滤的 WebSocket 服务。可以按 collection 类型或用户 DID 过滤，从而只接收需要的事件，效率更高。

特性	说明
数据格式	JSON
过滤	可按 collection、DID 过滤
数据量	取决于过滤条件，可以较为轻量
用途	监听帖子、点赞、关注等

启动方式

# 接收全部消息（无过滤）
php artisan bluesky:ws start

# 调试：显示所有接收到的消息
php artisan bluesky:ws start -v

collection 过滤

通过 -C 选项过滤要接收的 collection。可以多次指定。

# 只接收帖子和点赞
php artisan bluesky:ws start -C app.bsky.feed.post -C app.bsky.feed.like

# 只接收关注事件
php artisan bluesky:ws start -C app.bsky.graph.follow

常用 collection:

collection	内容
`app.bsky.feed.post`	帖子的创建 / 删除
`app.bsky.feed.like`	点赞
`app.bsky.feed.repost`	转发
`app.bsky.graph.follow`	关注
`app.bsky.graph.block`	屏蔽

DID 过滤

通过 -D 选项只接收特定用户的事件。

# 只接收特定用户的帖子
php artisan bluesky:ws start -C app.bsky.feed.post -D did:plc:xxx -D did:plc:yyy

事件处理

Jetstream 命令会根据接收到的消息类型派发 Laravel 事件。

事件类	触发时机
`JetstreamMessageReceived`	收到任何消息时
`JetstreamCommitMessage`	record 的创建 / 更新 / 删除时
`JetstreamIdentityMessage`	handle 变更等 identity 事件时
`JetstreamAccountMessage`	账号启用 / 停用时

创建事件监听器来处理事件。

php artisan make:listener JetstreamPostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Jetstream\JetstreamCommitMessage;

class JetstreamPostListener
{
    public function handle(JetstreamCommitMessage $event): void
    {
        // 判断 collection 类型
        $collection = data_get($event->message, 'commit.collection');

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

        // 操作类型：create / update / delete
        $operation = $event->operation;

        // 发帖者 DID
        $did = $event->message['did'];

        // record 内容
        $record = data_get($event->message, 'commit.record');
        $text = data_get($record, 'text', '');

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

Firehose

概览

Firehose 是 AT Protocol 的原始事件流。可以以二进制（DAG-CBOR）格式接收 Bluesky 网络上所有的 record 操作。

特性	说明
数据格式	DAG-CBOR 二进制（包会自动解码）
过滤	无（全部接收）
数据量	非常庞大
用途	全量数据的收集与归档

DAG-CBOR 的解码由包自动完成。在事件监听器中，你可以像使用普通 PHP 数组一样处理数据。

启动方式

php artisan bluesky:firehose start

# 调试：显示接收到的消息
php artisan bluesky:firehose start -v

事件处理

Firehose 命令也是通过 Laravel 事件来处理消息。

事件类	触发时机
`FirehoseMessageReceived`	收到任何消息时（包含原始数据）
`FirehoseCommitMessage`	record 的创建 / 更新 / 删除时
`FirehoseIdentityMessage`	identity 事件时
`FirehoseAccountMessage`	账号事件时
`FirehoseSyncMessage`	仓库同步事件时

php artisan make:listener FirehosePostListener

namespace App\Listeners;

use Revolution\Bluesky\Events\Firehose\FirehoseCommitMessage;

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

        // 操作类型：create / update / delete
        $action = $event->action;

        // 发帖者 DID
        $did = $event->did;

        // record 内容（已解码为数组）
        $record = $event->record;
        $text = data_get($record, 'value.text', '');

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

配置

可以在 config/bluesky.php 中修改连接目标主机和日志配置。

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

与 Labeler 结合

可以同时启动 Labeler 服务器和 Jetstream / Firehose。Labeler 处理打 label 请求时可以利用 Jetstream 或 Firehose 的数据。

# Labeler + Jetstream（监听关注事件）
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose（接收全部数据）
php artisan bluesky:labeler:server start --firehose

Labeler 的详情请参考 Labeler 页面。

长时进程的运维

WebSocket 命令是长期运行的进程。在生产环境中，请使用 Supervisor 等进程管理工具。

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

# 重新加载 Supervisor 并启动
sudo supervisorctl reread
sudo supervisorctl update
sudo supervisorctl start bluesky-jetstream:*

在 Laravel Forge 中配置守护进程

如果你使用 Laravel Forge，可以在 Daemons 部分添加守护进程。

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

在 Laravel Cloud 中配置后台进程

WebSocket 命令作为 WebSocket 客户端 连接到 Bluesky 的事件流，因此在 Laravel Cloud 上也能运行。将其作为 Laravel Cloud 的后台进程（自定义 worker）配置即可。在 Laravel Cloud 的后台进程配置中添加 Custom Worker。 Jetstream：

php artisan bluesky:ws start

Firehose：

php artisan bluesky:firehose start

部署时的进程停止与重启等操作，都会由 Laravel Cloud 自动处理。除了后台进程本身的配置外，不需要其他额外设置。

注意事项

进程意外退出时，autorestart=true 会自动将其重启。
为了避免内存泄漏，可以考虑定期重启。
对于消息量巨大的 Firehose，建议把 listener 中的处理异步化（通过 Queue Job）。

// 在 listener 中将处理派发给 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;
        }

        // 将耗时处理委托给 Queue Job
        ProcessFirehosePost::dispatch($event->did, $event->record, $event->action);
    }
}

Source: src/Console/WebSocket

​概览

​安装

​Jetstream

​概览

​启动方式

​collection 过滤

​DID 过滤

​事件处理

​Firehose

​概览

​启动方式

​事件处理

​配置

​与 Labeler 结合

​长时进程的运维

​Supervisor 配置示例

​在 Laravel Forge 中配置守护进程

​在 Laravel Cloud 中配置后台进程

​注意事项

概览

安装

Jetstream

概览

启动方式

collection 过滤

DID 过滤

事件处理

Firehose

概览

启动方式

事件处理

配置

与 Labeler 结合

长时进程的运维

Supervisor 配置示例

在 Laravel Forge 中配置守护进程

在 Laravel Cloud 中配置后台进程

注意事项