laravel-bluesky는 Bluesky의 실시간 스트림에 접속하기 위한 2가지 WebSocket 명령어를 제공합니다.
- Jetstream — Bluesky 고유의 필터링된 WebSocket 엔드포인트. JSON 형식으로 경량.
- Firehose — AT Protocol의 원본 이벤트 스트림. DAG-CBOR 바이너리 형식으로 모든 데이터를 수신.
WebSocket에 의한 장시간 실행 프로세스는 VPS나 EC2 등의 상시 기동 서버 또는 Laravel Cloud의 커스텀 워커가 필요합니다. Laravel Vapor나 Vercel 등의 서버리스 환경에서는 동작하지 않습니다.
composer require workerman/workerman
Jetstream
Jetstream은 Bluesky가 제공하는 필터링된 WebSocket 서비스입니다. 컬렉션 종류나 사용자 DID로 필터링할 수 있어, 필요한 이벤트만 효율적으로 수신할 수 있습니다.
| 특징 | 설명 |
|---|
| 데이터 형식 | JSON |
| 필터링 | 컬렉션·DID로 필터 가능 |
| 데이터량 | 필터에 따라 경량 |
| 용도 | 게시·좋아요·팔로우 등 감시 |
기동 방법
# 모든 메시지 수신(필터 없음)
php artisan bluesky:ws start
# 디버그: 수신한 모든 메시지를 표시
php artisan bluesky:ws start -v
컬렉션 필터
-C 옵션으로 수신할 컬렉션을 좁힙니다. 여러 개 지정할 수 있습니다.
# 게시와 좋아요만 수신
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
| 컬렉션 | 내용 |
|---|
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 | 레코드 작성·갱신·삭제 시 |
JetstreamIdentityMessage | 핸들 변경 등 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 = 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 = data_get($event->message, 'commit.record');
$text = data_get($record, 'text', '');
info("[$operation] $did: $text");
}
}
Firehose
Firehose는 AT Protocol의 원본 이벤트 스트림입니다. Bluesky 네트워크상의 모든 레코드 조작을 바이너리(DAG-CBOR) 형식으로 수신합니다.
| 특징 | 설명 |
|---|
| 데이터 형식 | DAG-CBOR 바이너리(패키지가 자동 디코드) |
| 필터링 | 없음(모든 데이터 수신) |
| 데이터량 | 매우 대량 |
| 용도 | 모든 데이터 수집·아카이브 |
DAG-CBOR 디코드는 패키지가 자동으로 수행합니다. 이벤트 리스너에서는 일반 PHP 배열로 데이터를 받을 수 있습니다.
기동 방법
php artisan bluesky:firehose start
# 디버그: 수신한 메시지 표시
php artisan bluesky:firehose start -v
이벤트 처리
Firehose 명령어도 Laravel 이벤트를 사용해 메시지를 처리합니다.
| 이벤트 클래스 | 타이밍 |
|---|
FirehoseMessageReceived | 모든 메시지 수신 시(원본 데이터 포함) |
FirehoseCommitMessage | 레코드 작성·갱신·삭제 시 |
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
{
// 컬렉션 종류 확인
if ($event->collection !== 'app.bsky.feed.post') {
return;
}
// 조작 종류: create / update / delete
$action = $event->action;
// 게시자의 DID
$did = $event->did;
// 레코드 내용(디코드된 배열)
$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가 받은 라벨링 요청 처리에 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
장시간 실행 프로세스의 운용
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의 백그라운드 프로세스(커스텀 워커)로 설정하세요.
Laravel Cloud의 백그라운드 프로세스 설정에서 Custom Worker를 추가합니다.
Jetstream의 경우:
php artisan bluesky:ws start
php artisan bluesky:firehose start
배포 시 프로세스 정지·재기동 등은 모두 Laravel Cloud 측에서 자동으로 처리됩니다. 백그라운드 프로세스 설정 외에 추가 설정은 불필요합니다.
주의 사항
- 프로세스가 예기치 못하게 종료된 경우,
autorestart=true로 자동 재기동됩니다.
- 메모리 누수를 방지하기 위해 정기적인 재기동을 검토해 주세요.
- 대량의 메시지를 수신하는 Firehose에서는 리스너 내부의 처리를 비동기(Queue Job)로 하는 것을 권장합니다.
// 리스너에서 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);
}
}
