> ## Documentation Index
> Fetch the complete documentation index at: https://kawax.biz/llms.txt
> Use this file to discover all available pages before exploring further.

# WebSocket (Jetstream / Firehose)

> Laravel Bluesky의 WebSocket 기능. Bluesky Jetstream과 AT Protocol Firehose를 사용한 실시간 이벤트 처리 방법. JetstreamServeCommand와 FirehoseServeCommand의 사용법을 설명합니다.

## 개요

`laravel-bluesky`는 Bluesky의 실시간 스트림에 접속하기 위한 2가지 WebSocket 명령어를 제공합니다.

* **Jetstream** — Bluesky 고유의 필터링된 WebSocket 엔드포인트. JSON 형식으로 경량.
* **Firehose** — AT Protocol의 원본 이벤트 스트림. DAG-CBOR 바이너리 형식으로 모든 데이터를 수신.

```mermaid theme={null}
graph LR
    subgraph "Bluesky 네트워크"
        J["Jetstream<br>jetstream1.us-west.bsky.network<br>JSON / 필터링 가능"]
        F["Firehose<br>bsky.network<br>DAG-CBOR / 모든 데이터"]
    end
    subgraph "Laravel 앱"
        WS["JetstreamServeCommand<br>FirehoseServeCommand<br>(Workerman)"]
        E["Laravel Events<br>JetstreamCommitMessage<br>FirehoseCommitMessage 등"]
        L["Listener / Job"]
    end

    J -->|WebSocket| WS
    F -->|WebSocket| WS
    WS --> E --> L
```

<Warning>
  WebSocket에 의한 장시간 실행 프로세스는 **VPS나 EC2 등의 상시 기동 서버** 또는 **Laravel Cloud의 커스텀 워커**가 필요합니다. Laravel Vapor나 Vercel 등의 서버리스 환경에서는 동작하지 않습니다.
</Warning>

## 설치

WebSocket 기능에는 [Workerman](https://github.com/walkor/workerman)이 필요합니다.

```bash theme={null}
composer require workerman/workerman
```

## Jetstream

### 개요

Jetstream은 Bluesky가 제공하는 필터링된 WebSocket 서비스입니다. 컬렉션 종류나 사용자 DID로 필터링할 수 있어, 필요한 이벤트만 효율적으로 수신할 수 있습니다.

| 특징     | 설명              |
| ------ | --------------- |
| 데이터 형식 | JSON            |
| 필터링    | 컬렉션·DID로 필터 가능  |
| 데이터량   | 필터에 따라 경량       |
| 용도     | 게시·좋아요·팔로우 등 감시 |

### 기동 방법

```bash theme={null}
# 모든 메시지 수신(필터 없음)
php artisan bluesky:ws start

# 디버그: 수신한 모든 메시지를 표시
php artisan bluesky:ws start -v
```

### 컬렉션 필터

`-C` 옵션으로 수신할 컬렉션을 좁힙니다. 여러 개 지정할 수 있습니다.

```bash theme={null}
# 게시와 좋아요만 수신
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` 옵션으로 특정 사용자의 이벤트만 수신합니다.

```bash theme={null}
# 특정 사용자의 게시만 수신
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`  | 계정 활성화·비활성화 시          |

이벤트 리스너를 만들어 이벤트를 처리합니다.

```bash theme={null}
php artisan make:listener JetstreamPostListener
```

```php theme={null}
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 바이너리(패키지가 자동 디코드) |
| 필터링    | 없음(모든 데이터 수신)              |
| 데이터량   | 매우 대량                      |
| 용도     | 모든 데이터 수집·아카이브             |

<Info>
  DAG-CBOR 디코드는 패키지가 자동으로 수행합니다. 이벤트 리스너에서는 일반 PHP 배열로 데이터를 받을 수 있습니다.
</Info>

### 기동 방법

```bash theme={null}
php artisan bluesky:firehose start

# 디버그: 수신한 메시지 표시
php artisan bluesky:firehose start -v
```

### 이벤트 처리

Firehose 명령어도 Laravel 이벤트를 사용해 메시지를 처리합니다.

| 이벤트 클래스                   | 타이밍                    |
| ------------------------- | ---------------------- |
| `FirehoseMessageReceived` | 모든 메시지 수신 시(원본 데이터 포함) |
| `FirehoseCommitMessage`   | 레코드 작성·갱신·삭제 시         |
| `FirehoseIdentityMessage` | identity 이벤트 시         |
| `FirehoseAccountMessage`  | 계정 이벤트 시               |
| `FirehoseSyncMessage`     | 저장소 동기화 이벤트 시          |

```bash theme={null}
php artisan make:listener FirehosePostListener
```

```php theme={null}
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`에서 접속 호스트나 로그 설정을 변경할 수 있습니다.

```php theme={null}
// 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`에서의 설정 예시:

```ini theme={null}
BLUESKY_JETSTREAM_HOST=jetstream2.us-east.bsky.network
BLUESKY_JETSTREAM_MAX=1000000
```

## Labeler와의 조합

Labeler 서버와 Jetstream / Firehose를 동시에 기동할 수 있습니다. Labeler가 받은 라벨링 요청 처리에 Jetstream이나 Firehose의 데이터를 이용할 수 있습니다.

```bash theme={null}
# Labeler + Jetstream(팔로우 이벤트를 감시)
php artisan bluesky:labeler:server start --jetstream -C app.bsky.graph.follow

# Labeler + Firehose(모든 데이터를 수신)
php artisan bluesky:labeler:server start --firehose
```

<Info>
  Labeler의 상세는 [Labeler 페이지](/ko/packages/laravel-bluesky/labeler)를 참조하세요.
</Info>

## 장시간 실행 프로세스의 운용

WebSocket 명령어는 장시간 기동을 지속하는 프로세스입니다. 프로덕션 환경에서는 Supervisor 등의 프로세스 관리 툴을 사용하세요.

### Supervisor 설정 예시

`/etc/supervisor/conf.d/bluesky-jetstream.conf`:

```ini theme={null}
[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
```

```bash theme={null}
# 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의 경우:**

```bash theme={null}
php artisan bluesky:ws start
```

**Firehose의 경우:**

```bash theme={null}
php artisan bluesky:firehose start
```

<Info>
  배포 시 프로세스 정지·재기동 등은 모두 Laravel Cloud 측에서 자동으로 처리됩니다. 백그라운드 프로세스 설정 외에 추가 설정은 불필요합니다.
</Info>

### 주의 사항

* 프로세스가 예기치 못하게 종료된 경우, `autorestart=true`로 자동 재기동됩니다.
* 메모리 누수를 방지하기 위해 정기적인 재기동을 검토해 주세요.
* 대량의 메시지를 수신하는 Firehose에서는 리스너 내부의 처리를 비동기(Queue Job)로 하는 것을 권장합니다.

```php theme={null}
// 리스너에서 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);
    }
}
```

<Info>
  Source: [src/Console/WebSocket](https://github.com/invokable/laravel-bluesky/tree/main/src/Console/WebSocket)
</Info>


## Related topics

- [Laravel Cloud — Laravel 전용 PaaS의 전모](/ko/blog/laravel-cloud.md)
- [스트리밍](/ko/packages/laravel-copilot-sdk/streaming.md)
- [Laravel Nostr](/ko/packages/laravel-nostr.md)
- [Laravel Reverb](/ko/reverb.md)
- [Laravel Notification for Discord(Webhook)](/ko/packages/laravel-notification-discord-webhook.md)