Build Reactive Signals for Bluesky's AT Protocol Firehose in Laravel
Implement core functionality
+2415
-204
+21
LICENSE
+21
LICENSE
···
1
+
MIT License
2
+
3
+
Copyright (c) 2025 Social Dept.
4
+
5
+
Permission is hereby granted, free of charge, to any person obtaining a copy
6
+
of this software and associated documentation files (the "Software"), to deal
7
+
in the Software without restriction, including without limitation the rights
8
+
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
9
+
copies of the Software, and to permit persons to whom the Software is
10
+
furnished to do so, subject to the following conditions:
11
+
12
+
The above copyright notice and this permission notice shall be included in all
13
+
copies or substantial portions of the Software.
14
+
15
+
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
16
+
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
17
+
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
18
+
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
19
+
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
20
+
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
21
+
SOFTWARE.
+618
README.md
+618
README.md
···
1
+
# Signal
2
+
3
+
**Laravel package for building Signals that respond to AT Protocol Jetstream events**
4
+
5
+
Signal provides a clean, Laravel-style interface for consuming real-time events from the AT Protocol firehose (Jetstream). Build reactive applications that respond to posts, likes, follows, and other social interactions on the AT Protocol network.
6
+
7
+
---
8
+
9
+
## Features
10
+
11
+
- 🔌 **WebSocket Connection** - Connect to AT Protocol Jetstream with automatic reconnection
12
+
- 🎯 **Signal-based Architecture** - Clean, testable event handlers (avoiding Laravel's "listener" naming collision)
13
+
- ⭐ **Wildcard Collection Filtering** - Match multiple collections with patterns like `app.bsky.feed.*`
14
+
- 💾 **Cursor Management** - Resume from last position after disconnections (Database, Redis, or File storage)
15
+
- ⚡ **Queue Integration** - Process events asynchronously with Laravel queues
16
+
- 🔍 **Auto-Discovery** - Automatically find and register Signals in `app/Signals`
17
+
- 🧪 **Testing Tools** - Test your Signals with sample data
18
+
- 🛠️ **Artisan Commands** - Full CLI support for managing and testing Signals
19
+
20
+
---
21
+
22
+
## Table of Contents
23
+
24
+
- [Installation](#installation)
25
+
- [Quick Start](#quick-start)
26
+
- [Creating Signals](#creating-signals)
27
+
- [Filtering Events](#filtering-events)
28
+
- [Queue Integration](#queue-integration)
29
+
- [Configuration](#configuration)
30
+
- [Available Commands](#available-commands)
31
+
- [Testing](#testing)
32
+
- [Documentation](#documentation)
33
+
- [License](#license)
34
+
35
+
---
36
+
37
+
## Installation
38
+
39
+
Install the package via Composer:
40
+
41
+
```bash
42
+
composer require social-dept/signal
43
+
```
44
+
45
+
Run the installation command:
46
+
47
+
```bash
48
+
php artisan signal:install
49
+
```
50
+
51
+
This will:
52
+
- Publish the configuration file to `config/signal.php`
53
+
- Publish the database migration
54
+
- Run migrations (with confirmation)
55
+
- Display next steps
56
+
57
+
### Manual Installation
58
+
59
+
If you prefer manual installation:
60
+
61
+
```bash
62
+
php artisan vendor:publish --tag=signal-config
63
+
php artisan vendor:publish --tag=signal-migrations
64
+
php artisan migrate
65
+
```
66
+
67
+
---
68
+
69
+
## Quick Start
70
+
71
+
### 1. Create Your First Signal
72
+
73
+
```bash
74
+
php artisan make:signal NewPostSignal
75
+
```
76
+
77
+
This creates `app/Signals/NewPostSignal.php`:
78
+
79
+
```php
80
+
<?php
81
+
82
+
namespace App\Signals;
83
+
84
+
use SocialDept\Signal\Events\JetstreamEvent;
85
+
use SocialDept\Signal\Signals\Signal;
86
+
87
+
class NewPostSignal extends Signal
88
+
{
89
+
public function eventTypes(): array
90
+
{
91
+
return ['commit'];
92
+
}
93
+
94
+
public function collections(): ?array
95
+
{
96
+
return ['app.bsky.feed.post'];
97
+
}
98
+
99
+
public function handle(JetstreamEvent $event): void
100
+
{
101
+
$record = $event->getRecord();
102
+
103
+
logger()->info('New post created', [
104
+
'did' => $event->did,
105
+
'text' => $record->text ?? null,
106
+
]);
107
+
}
108
+
}
109
+
```
110
+
111
+
### 2. Start Consuming Events
112
+
113
+
```bash
114
+
php artisan signal:consume
115
+
```
116
+
117
+
Your Signal will now respond to new posts on the AT Protocol network in real-time!
118
+
119
+
---
120
+
121
+
## Creating Signals
122
+
123
+
### Basic Signal Structure
124
+
125
+
Every Signal extends the base `Signal` class and must implement:
126
+
127
+
```php
128
+
use SocialDept\Signal\Events\JetstreamEvent;
129
+
use SocialDept\Signal\Signals\Signal;
130
+
131
+
class MySignal extends Signal
132
+
{
133
+
// Required: Define which event types to listen for
134
+
public function eventTypes(): array
135
+
{
136
+
return ['commit']; // 'commit', 'identity', or 'account'
137
+
}
138
+
139
+
// Required: Handle the event
140
+
public function handle(JetstreamEvent $event): void
141
+
{
142
+
// Your logic here
143
+
}
144
+
}
145
+
```
146
+
147
+
### Event Types
148
+
149
+
Three event types are available:
150
+
151
+
| Type | Description | Use Cases |
152
+
|------|-------------|-----------|
153
+
| `commit` | Repository commits (posts, likes, follows, etc.) | Content creation, social interactions |
154
+
| `identity` | Identity changes (handle updates) | User profile tracking |
155
+
| `account` | Account status changes | Account monitoring |
156
+
157
+
### Accessing Event Data
158
+
159
+
```php
160
+
public function handle(JetstreamEvent $event): void
161
+
{
162
+
// Common properties
163
+
$did = $event->did; // User's DID
164
+
$kind = $event->kind; // Event type
165
+
$timestamp = $event->timeUs; // Microsecond timestamp
166
+
167
+
// Commit events
168
+
if ($event->isCommit()) {
169
+
$collection = $event->getCollection(); // e.g., 'app.bsky.feed.post'
170
+
$operation = $event->getOperation(); // 'create', 'update', or 'delete'
171
+
$record = $event->getRecord(); // The actual record data
172
+
$rkey = $event->commit->rkey; // Record key
173
+
}
174
+
175
+
// Identity events
176
+
if ($event->isIdentity()) {
177
+
$handle = $event->identity->handle;
178
+
}
179
+
180
+
// Account events
181
+
if ($event->isAccount()) {
182
+
$active = $event->account->active;
183
+
$status = $event->account->status;
184
+
}
185
+
}
186
+
```
187
+
188
+
---
189
+
190
+
## Filtering Events
191
+
192
+
### Collection Filtering (with Wildcards!)
193
+
194
+
Filter events by AT Protocol collection:
195
+
196
+
```php
197
+
// Exact match - only posts
198
+
public function collections(): ?array
199
+
{
200
+
return ['app.bsky.feed.post'];
201
+
}
202
+
203
+
// Wildcard - all feed events
204
+
public function collections(): ?array
205
+
{
206
+
return ['app.bsky.feed.*'];
207
+
}
208
+
209
+
// Multiple patterns
210
+
public function collections(): ?array
211
+
{
212
+
return [
213
+
'app.bsky.feed.post',
214
+
'app.bsky.feed.repost',
215
+
'app.bsky.graph.*', // All graph collections
216
+
];
217
+
}
218
+
219
+
// No filter - all collections
220
+
public function collections(): ?array
221
+
{
222
+
return null;
223
+
}
224
+
```
225
+
226
+
### Common Collection Patterns
227
+
228
+
| Pattern | Matches |
229
+
|---------|---------|
230
+
| `app.bsky.feed.*` | Posts, likes, reposts, etc. |
231
+
| `app.bsky.graph.*` | Follows, blocks, mutes |
232
+
| `app.bsky.actor.*` | Profile updates |
233
+
| `app.bsky.*` | All Bluesky collections |
234
+
235
+
### DID Filtering
236
+
237
+
Filter events by specific users:
238
+
239
+
```php
240
+
public function dids(): ?array
241
+
{
242
+
return [
243
+
'did:plc:z72i7hdynmk6r22z27h6tvur', // Specific user
244
+
'did:plc:ragtjsm2j2vknwkz3zp4oxrd', // Another user
245
+
];
246
+
}
247
+
```
248
+
249
+
### Custom Filtering
250
+
251
+
Add complex filtering logic:
252
+
253
+
```php
254
+
public function shouldHandle(JetstreamEvent $event): bool
255
+
{
256
+
// Only handle posts with images
257
+
if ($event->isCommit() && $event->commit->collection === 'app.bsky.feed.post') {
258
+
$record = $event->getRecord();
259
+
return isset($record->embed);
260
+
}
261
+
262
+
return true;
263
+
}
264
+
```
265
+
266
+
---
267
+
268
+
## Queue Integration
269
+
270
+
Process events asynchronously using Laravel queues:
271
+
272
+
```php
273
+
class HeavyProcessingSignal extends Signal
274
+
{
275
+
public function eventTypes(): array
276
+
{
277
+
return ['commit'];
278
+
}
279
+
280
+
// Enable queueing
281
+
public function shouldQueue(): bool
282
+
{
283
+
return true;
284
+
}
285
+
286
+
// Optional: Customize queue
287
+
public function queue(): string
288
+
{
289
+
return 'high-priority';
290
+
}
291
+
292
+
// Optional: Customize connection
293
+
public function queueConnection(): string
294
+
{
295
+
return 'redis';
296
+
}
297
+
298
+
public function handle(JetstreamEvent $event): void
299
+
{
300
+
// This runs in a queue job
301
+
$this->performExpensiveOperation($event);
302
+
}
303
+
304
+
// Handle failures
305
+
public function failed(JetstreamEvent $event, \Throwable $exception): void
306
+
{
307
+
Log::error('Signal failed', [
308
+
'event' => $event->toArray(),
309
+
'error' => $exception->getMessage(),
310
+
]);
311
+
}
312
+
}
313
+
```
314
+
315
+
---
316
+
317
+
## Configuration
318
+
319
+
Configuration is stored in `config/signal.php`:
320
+
321
+
### Jetstream URL
322
+
323
+
```php
324
+
'websocket_url' => env('SIGNAL_JETSTREAM_URL', 'wss://jetstream2.us-east.bsky.network'),
325
+
```
326
+
327
+
Available endpoints:
328
+
- **US East**: `wss://jetstream2.us-east.bsky.network`
329
+
- **US West**: `wss://jetstream1.us-west.bsky.network`
330
+
331
+
### Cursor Storage
332
+
333
+
Choose how to store cursor positions:
334
+
335
+
```php
336
+
'cursor_storage' => env('SIGNAL_CURSOR_STORAGE', 'database'),
337
+
```
338
+
339
+
| Driver | Best For | Configuration |
340
+
|--------|----------|---------------|
341
+
| `database` | Production, multi-server | Default connection |
342
+
| `redis` | High performance, distributed | Redis connection |
343
+
| `file` | Development, single server | Storage path |
344
+
345
+
### Environment Variables
346
+
347
+
Add to your `.env`:
348
+
349
+
```env
350
+
# Required
351
+
SIGNAL_JETSTREAM_URL=wss://jetstream2.us-east.bsky.network
352
+
353
+
# Optional
354
+
SIGNAL_CURSOR_STORAGE=database
355
+
SIGNAL_QUEUE_CONNECTION=redis
356
+
SIGNAL_QUEUE=signal
357
+
SIGNAL_BATCH_SIZE=100
358
+
SIGNAL_RATE_LIMIT=1000
359
+
```
360
+
361
+
### Auto-Discovery
362
+
363
+
Signals are automatically discovered from `app/Signals`. Disable if needed:
364
+
365
+
```php
366
+
'auto_discovery' => [
367
+
'enabled' => true,
368
+
'path' => app_path('Signals'),
369
+
'namespace' => 'App\\Signals',
370
+
],
371
+
```
372
+
373
+
Or manually register Signals:
374
+
375
+
```php
376
+
'signals' => [
377
+
\App\Signals\NewPostSignal::class,
378
+
\App\Signals\NewFollowSignal::class,
379
+
],
380
+
```
381
+
382
+
---
383
+
384
+
## Available Commands
385
+
386
+
### `signal:install`
387
+
Install the package (publish config, migrations, run migrations)
388
+
389
+
```bash
390
+
php artisan signal:install
391
+
```
392
+
393
+
### `signal:consume`
394
+
Start consuming events from Jetstream
395
+
396
+
```bash
397
+
php artisan signal:consume
398
+
399
+
# Start from specific cursor
400
+
php artisan signal:consume --cursor=123456789
401
+
402
+
# Start fresh (ignore stored cursor)
403
+
php artisan signal:consume --fresh
404
+
```
405
+
406
+
### `signal:list`
407
+
List all registered Signals
408
+
409
+
```bash
410
+
php artisan signal:list
411
+
```
412
+
413
+
### `signal:make`
414
+
Create a new Signal class
415
+
416
+
```bash
417
+
php artisan make:signal NewPostSignal
418
+
419
+
# With options
420
+
php artisan make:signal FollowSignal --type=commit --collection=app.bsky.graph.follow
421
+
```
422
+
423
+
### `signal:test`
424
+
Test a Signal with sample data
425
+
426
+
```bash
427
+
php artisan signal:test NewPostSignal
428
+
```
429
+
430
+
---
431
+
432
+
## Testing
433
+
434
+
Signal includes a comprehensive test suite. Test your Signals:
435
+
436
+
### Unit Testing
437
+
438
+
```php
439
+
use SocialDept\Signal\Events\CommitEvent;
440
+
use SocialDept\Signal\Events\JetstreamEvent;
441
+
442
+
class NewPostSignalTest extends TestCase
443
+
{
444
+
/** @test */
445
+
public function it_handles_new_posts()
446
+
{
447
+
$signal = new NewPostSignal();
448
+
449
+
$event = new JetstreamEvent(
450
+
did: 'did:plc:test',
451
+
timeUs: time() * 1000000,
452
+
kind: 'commit',
453
+
commit: new CommitEvent(
454
+
rev: 'test',
455
+
operation: 'create',
456
+
collection: 'app.bsky.feed.post',
457
+
rkey: 'test',
458
+
record: (object) [
459
+
'text' => 'Hello World!',
460
+
'createdAt' => now()->toIso8601String(),
461
+
],
462
+
),
463
+
);
464
+
465
+
$signal->handle($event);
466
+
467
+
// Assert your expected behavior
468
+
}
469
+
}
470
+
```
471
+
472
+
### Testing with Artisan
473
+
474
+
```bash
475
+
php artisan signal:test NewPostSignal
476
+
```
477
+
478
+
---
479
+
480
+
## Documentation
481
+
482
+
For detailed documentation, see:
483
+
484
+
- **[INSTALLATION.md](./INSTALLATION.md)** - Complete installation guide with troubleshooting
485
+
- **[PACKAGE_SUMMARY.md](./PACKAGE_SUMMARY.md)** - Quick reference for package components
486
+
- **[WILDCARD_EXAMPLES.md](./WILDCARD_EXAMPLES.md)** - Comprehensive wildcard pattern guide
487
+
- **[IMPLEMENTATION_PLAN.md](./IMPLEMENTATION_PLAN.md)** - Full architecture and implementation details
488
+
489
+
### External Resources
490
+
491
+
- [AT Protocol Documentation](https://atproto.com/)
492
+
- [Jetstream Documentation](https://docs.bsky.app/docs/advanced-guides/jetstream)
493
+
- [Bluesky Lexicon](https://atproto.com/lexicons)
494
+
495
+
---
496
+
497
+
## Examples
498
+
499
+
### Monitor All Feed Activity
500
+
501
+
```php
502
+
class FeedMonitorSignal extends Signal
503
+
{
504
+
public function eventTypes(): array
505
+
{
506
+
return ['commit'];
507
+
}
508
+
509
+
public function collections(): ?array
510
+
{
511
+
return ['app.bsky.feed.*'];
512
+
}
513
+
514
+
public function handle(JetstreamEvent $event): void
515
+
{
516
+
// Handles posts, likes, reposts, etc.
517
+
Log::info('Feed activity', [
518
+
'collection' => $event->getCollection(),
519
+
'operation' => $event->getOperation(),
520
+
'did' => $event->did,
521
+
]);
522
+
}
523
+
}
524
+
```
525
+
526
+
### Track New Follows
527
+
528
+
```php
529
+
class NewFollowSignal extends Signal
530
+
{
531
+
public function eventTypes(): array
532
+
{
533
+
return ['commit'];
534
+
}
535
+
536
+
public function collections(): ?array
537
+
{
538
+
return ['app.bsky.graph.follow'];
539
+
}
540
+
541
+
public function handle(JetstreamEvent $event): void
542
+
{
543
+
if ($event->commit->isCreate()) {
544
+
$record = $event->getRecord();
545
+
546
+
// Store follow relationship
547
+
Follow::create([
548
+
'follower_did' => $event->did,
549
+
'following_did' => $record->subject,
550
+
]);
551
+
}
552
+
}
553
+
}
554
+
```
555
+
556
+
### Content Moderation
557
+
558
+
```php
559
+
class ModerationSignal extends Signal
560
+
{
561
+
public function eventTypes(): array
562
+
{
563
+
return ['commit'];
564
+
}
565
+
566
+
public function collections(): ?array
567
+
{
568
+
return ['app.bsky.feed.post'];
569
+
}
570
+
571
+
public function shouldQueue(): bool
572
+
{
573
+
return true;
574
+
}
575
+
576
+
public function handle(JetstreamEvent $event): void
577
+
{
578
+
$record = $event->getRecord();
579
+
580
+
if ($this->containsProhibitedContent($record->text)) {
581
+
$this->flagForModeration($event->did, $record);
582
+
}
583
+
}
584
+
}
585
+
```
586
+
587
+
---
588
+
589
+
## Requirements
590
+
591
+
- PHP 8.2 or higher
592
+
- Laravel 11.0 or higher
593
+
- WebSocket support (enabled by default in most environments)
594
+
595
+
---
596
+
597
+
## License
598
+
599
+
The MIT License (MIT). Please see [LICENSE](LICENSE) for more information.
600
+
601
+
---
602
+
603
+
## Contributing
604
+
605
+
Contributions are welcome! Please see [CONTRIBUTING.md](contributing.md) for details.
606
+
607
+
---
608
+
609
+
## Support
610
+
611
+
For issues, questions, or feature requests:
612
+
- Open an issue on GitHub
613
+
- Check the [documentation files](#documentation)
614
+
- Review the [implementation plan](./IMPLEMENTATION_PLAN.md)
615
+
616
+
---
617
+
618
+
**Built for the AT Protocol ecosystem** • Made with ❤️ by Social Dept
-8
changelog.md
-8
changelog.md
+11
-14
composer.json
+11
-14
composer.json
···
1
1
{
2
2
"name": "social-dept/signal",
3
-
"description": ":package_description",
3
+
"description": "Laravel package for building Signals that respond to AT Protocol Jetstream events",
4
+
"type": "library",
4
5
"license": "MIT",
5
-
"authors": [
6
-
{
7
-
"name": "Author Name",
8
-
"email": "author@email.com",
9
-
"homepage": "http://author.com"
10
-
}
11
-
],
12
-
"homepage": "https://github.com/social-dept/signal",
13
-
"keywords": ["Laravel", "Signal"],
14
6
"require": {
15
-
"illuminate/support": "~9"
7
+
"php": "^8.2",
8
+
"illuminate/support": "^11.0",
9
+
"illuminate/console": "^11.0",
10
+
"illuminate/database": "^11.0",
11
+
"ratchet/pawl": "^0.4",
12
+
"react/event-loop": "^1.5"
16
13
},
17
14
"require-dev": {
18
-
"phpunit/phpunit": "~9.0",
19
-
"orchestra/testbench": "~7"
15
+
"orchestra/testbench": "^9.0",
16
+
"phpunit/phpunit": "^11.0"
20
17
},
21
18
"autoload": {
22
19
"psr-4": {
···
25
22
},
26
23
"autoload-dev": {
27
24
"psr-4": {
28
-
"SocialDept\\Signal\\Tests\\": "tests"
25
+
"SocialDept\\Signal\\Tests\\": "tests/"
29
26
}
30
27
},
31
28
"extra": {
+114
-2
config/signal.php
+114
-2
config/signal.php
···
1
1
<?php
2
2
3
3
return [
4
-
//
5
-
];
4
+
/*
5
+
|--------------------------------------------------------------------------
6
+
| Jetstream WebSocket URL
7
+
|--------------------------------------------------------------------------
8
+
|
9
+
| The WebSocket URL for the AT Protocol Jetstream service.
10
+
| US East: wss://jetstream2.us-east.bsky.network
11
+
| US West: wss://jetstream1.us-west.bsky.network
12
+
|
13
+
*/
14
+
'websocket_url' => env('SIGNAL_JETSTREAM_URL', 'wss://jetstream2.us-east.bsky.network'),
15
+
16
+
/*
17
+
|--------------------------------------------------------------------------
18
+
| Cursor Storage Driver
19
+
|--------------------------------------------------------------------------
20
+
|
21
+
| Determines how Signal stores the cursor position for resuming after
22
+
| disconnections. Options: 'database', 'redis', 'file'
23
+
|
24
+
*/
25
+
'cursor_storage' => env('SIGNAL_CURSOR_STORAGE', 'database'),
26
+
27
+
/*
28
+
|--------------------------------------------------------------------------
29
+
| Cursor Storage Configuration
30
+
|--------------------------------------------------------------------------
31
+
*/
32
+
'cursor_config' => [
33
+
'database' => [
34
+
'table' => 'signal_cursors',
35
+
'connection' => null, // null = default connection
36
+
],
37
+
'redis' => [
38
+
'connection' => 'default',
39
+
'key' => 'signal:cursor',
40
+
],
41
+
'file' => [
42
+
'path' => storage_path('signal/cursor.json'),
43
+
],
44
+
],
45
+
46
+
/*
47
+
|--------------------------------------------------------------------------
48
+
| Signals
49
+
|--------------------------------------------------------------------------
50
+
|
51
+
| Register your Signals here, or use auto-discovery by placing them
52
+
| in app/Signals directory.
53
+
|
54
+
*/
55
+
'signals' => [
56
+
// App\Signals\NewPostSignal::class,
57
+
],
58
+
59
+
/*
60
+
|--------------------------------------------------------------------------
61
+
| Auto-Discovery
62
+
|--------------------------------------------------------------------------
63
+
|
64
+
| Automatically discover Signals in the specified directory.
65
+
|
66
+
*/
67
+
'auto_discovery' => [
68
+
'enabled' => true,
69
+
'path' => app_path('Signals'),
70
+
'namespace' => 'App\\Signals',
71
+
],
72
+
73
+
/*
74
+
|--------------------------------------------------------------------------
75
+
| Queue Configuration
76
+
|--------------------------------------------------------------------------
77
+
|
78
+
| Default queue settings for Signals that should be queued.
79
+
|
80
+
*/
81
+
'queue' => [
82
+
'connection' => env('SIGNAL_QUEUE_CONNECTION', 'default'),
83
+
'queue' => env('SIGNAL_QUEUE', 'signal'),
84
+
],
85
+
86
+
/*
87
+
|--------------------------------------------------------------------------
88
+
| Connection Settings
89
+
|--------------------------------------------------------------------------
90
+
*/
91
+
'connection' => [
92
+
'reconnect_attempts' => 5,
93
+
'reconnect_delay' => 5, // seconds
94
+
'ping_interval' => 30, // seconds
95
+
'timeout' => 60, // seconds
96
+
],
97
+
98
+
/*
99
+
|--------------------------------------------------------------------------
100
+
| Performance & Rate Limiting
101
+
|--------------------------------------------------------------------------
102
+
*/
103
+
'performance' => [
104
+
'batch_size' => env('SIGNAL_BATCH_SIZE', 100),
105
+
'rate_limit' => env('SIGNAL_RATE_LIMIT', 1000), // events per minute
106
+
],
107
+
108
+
/*
109
+
|--------------------------------------------------------------------------
110
+
| Logging
111
+
|--------------------------------------------------------------------------
112
+
*/
113
+
'logging' => [
114
+
'channel' => env('SIGNAL_LOG_CHANNEL', 'stack'),
115
+
'level' => env('SIGNAL_LOG_LEVEL', 'info'),
116
+
],
117
+
];
-27
contributing.md
-27
contributing.md
···
1
-
# Contributing
2
-
3
-
Contributions are welcome and will be fully credited.
4
-
5
-
Contributions are accepted via Pull Requests on [Github](https://github.com/social-dept/signal).
6
-
7
-
# Things you could do
8
-
If you want to contribute but do not know where to start, this list provides some starting points.
9
-
- Add license text
10
-
- Remove rewriteRules.php
11
-
- Set up TravisCI, StyleCI, ScrutinizerCI
12
-
- Write a comprehensive ReadMe
13
-
14
-
## Pull Requests
15
-
16
-
- **Add tests!** - Your patch won't be accepted if it doesn't have tests.
17
-
18
-
- **Document any change in behaviour** - Make sure the `readme.md` and any other relevant documentation are kept up-to-date.
19
-
20
-
- **Consider our release cycle** - We try to follow [SemVer v2.0.0](http://semver.org/). Randomly breaking public APIs is not an option.
21
-
22
-
- **One pull request per feature** - If you want to do more than one thing, send multiple pull requests.
23
-
24
-
- **Send coherent history** - Make sure each individual commit in your pull request is meaningful. If you had to make multiple intermediate commits while developing, please [squash them](http://www.git-scm.com/book/en/v2/Git-Tools-Rewriting-History#Changing-Multiple-Commit-Messages) before submitting.
25
-
26
-
27
-
**Happy coding**!
+25
database/migrations/2024_01_01_000000_create_signal_cursors_table.php
+25
database/migrations/2024_01_01_000000_create_signal_cursors_table.php
···
1
+
<?php
2
+
3
+
use Illuminate\Database\Migrations\Migration;
4
+
use Illuminate\Database\Schema\Blueprint;
5
+
use Illuminate\Support\Facades\Schema;
6
+
7
+
return new class extends Migration
8
+
{
9
+
public function up(): void
10
+
{
11
+
Schema::create('signal_cursors', function (Blueprint $table) {
12
+
$table->id();
13
+
$table->string('key')->unique();
14
+
$table->bigInteger('cursor');
15
+
$table->timestamps();
16
+
17
+
$table->index('key');
18
+
});
19
+
}
20
+
21
+
public function down(): void
22
+
{
23
+
Schema::dropIfExists('signal_cursors');
24
+
}
25
+
};
-5
license.md
-5
license.md
+12
-17
phpunit.xml
+12
-17
phpunit.xml
···
1
1
<?xml version="1.0" encoding="UTF-8"?>
2
-
<phpunit bootstrap="vendor/autoload.php"
3
-
backupGlobals="false"
4
-
backupStaticAttributes="false"
5
-
colors="true"
6
-
verbose="true"
7
-
convertErrorsToExceptions="true"
8
-
convertNoticesToExceptions="true"
9
-
convertWarningsToExceptions="true"
10
-
processIsolation="false"
11
-
stopOnFailure="false">
2
+
<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3
+
xsi:noNamespaceSchemaLocation="vendor/phpunit/phpunit/phpunit.xsd"
4
+
bootstrap="vendor/autoload.php"
5
+
colors="true">
12
6
<testsuites>
13
-
<testsuite name="Package">
14
-
<directory suffix=".php">./tests/</directory>
7
+
<testsuite name="Signal Test Suite">
8
+
<directory>tests</directory>
15
9
</testsuite>
16
10
</testsuites>
17
-
<filter>
18
-
<whitelist>
19
-
<directory>src/</directory>
20
-
</whitelist>
21
-
</filter>
11
+
<php>
12
+
<env name="APP_ENV" value="testing"/>
13
+
<env name="CACHE_DRIVER" value="array"/>
14
+
<env name="SESSION_DRIVER" value="array"/>
15
+
<env name="QUEUE_CONNECTION" value="sync"/>
16
+
</php>
22
17
</phpunit>
-57
readme.md
-57
readme.md
···
1
-
# Signal
2
-
3
-
[![Latest Version on Packagist][ico-version]][link-packagist]
4
-
[![Total Downloads][ico-downloads]][link-downloads]
5
-
[![Build Status][ico-travis]][link-travis]
6
-
[![StyleCI][ico-styleci]][link-styleci]
7
-
8
-
This is where your description should go. Take a look at [contributing.md](contributing.md) to see a to do list.
9
-
10
-
## Installation
11
-
12
-
Via Composer
13
-
14
-
```bash
15
-
composer require social-dept/signal
16
-
```
17
-
18
-
## Usage
19
-
20
-
## Change log
21
-
22
-
Please see the [changelog](changelog.md) for more information on what has changed recently.
23
-
24
-
## Testing
25
-
26
-
```bash
27
-
composer test
28
-
```
29
-
30
-
## Contributing
31
-
32
-
Please see [contributing.md](contributing.md) for details and a todolist.
33
-
34
-
## Security
35
-
36
-
If you discover any security related issues, please email author@email.com instead of using the issue tracker.
37
-
38
-
## Credits
39
-
40
-
- [Author Name][link-author]
41
-
- [All Contributors][link-contributors]
42
-
43
-
## License
44
-
45
-
MIT. Please see the [license file](license.md) for more information.
46
-
47
-
[ico-version]: https://img.shields.io/packagist/v/social-dept/signal.svg?style=flat-square
48
-
[ico-downloads]: https://img.shields.io/packagist/dt/social-dept/signal.svg?style=flat-square
49
-
[ico-travis]: https://img.shields.io/travis/social-dept/signal/master.svg?style=flat-square
50
-
[ico-styleci]: https://styleci.io/repos/12345678/shield
51
-
52
-
[link-packagist]: https://packagist.org/packages/social-dept/signal
53
-
[link-downloads]: https://packagist.org/packages/social-dept/signal
54
-
[link-travis]: https://travis-ci.org/social-dept/signal
55
-
[link-styleci]: https://styleci.io/repos/12345678
56
-
[link-author]: https://github.com/social-dept
57
-
[link-contributors]: ../../contributors
+67
src/Commands/ConsumeCommand.php
+67
src/Commands/ConsumeCommand.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Commands;
4
+
5
+
use Illuminate\Console\Command;
6
+
use SocialDept\Signal\Services\JetstreamConsumer;
7
+
use SocialDept\Signal\Services\SignalRegistry;
8
+
9
+
class ConsumeCommand extends Command
10
+
{
11
+
protected $signature = 'signal:consume
12
+
{--cursor= : Start from a specific cursor position}
13
+
{--fresh : Start from the beginning, ignoring stored cursor}';
14
+
15
+
protected $description = 'Start consuming events from the AT Protocol Jetstream';
16
+
17
+
public function handle(JetstreamConsumer $consumer, SignalRegistry $registry): int
18
+
{
19
+
$this->info('Signal: Initializing Jetstream consumer...');
20
+
21
+
// Discover signals
22
+
$registry->discover();
23
+
24
+
$signalCount = $registry->all()->count();
25
+
$this->info("Registered {$signalCount} signal(s)");
26
+
27
+
if ($signalCount === 0) {
28
+
$this->warn('No signals registered. Create signals in app/Signals or register them in config/signal.php');
29
+
return self::FAILURE;
30
+
}
31
+
32
+
// List registered signals
33
+
$this->table(
34
+
['Signal', 'Event Types', 'Collections'],
35
+
$registry->all()->map(function ($signal) {
36
+
return [
37
+
get_class($signal),
38
+
implode(', ', $signal->eventTypes()),
39
+
$signal->collections() ? implode(', ', $signal->collections()) : 'All',
40
+
];
41
+
})
42
+
);
43
+
44
+
// Determine cursor
45
+
$cursor = null;
46
+
if ($this->option('fresh')) {
47
+
$this->info('Starting fresh from the beginning');
48
+
} elseif ($this->option('cursor')) {
49
+
$cursor = (int) $this->option('cursor');
50
+
$this->info("Starting from cursor: {$cursor}");
51
+
} else {
52
+
$this->info('Resuming from stored cursor position');
53
+
}
54
+
55
+
// Start consuming
56
+
$this->info('Starting Jetstream consumer... Press Ctrl+C to stop.');
57
+
58
+
try {
59
+
$consumer->start($cursor);
60
+
} catch (\Exception $e) {
61
+
$this->error('Error: ' . $e->getMessage());
62
+
return self::FAILURE;
63
+
}
64
+
65
+
return self::SUCCESS;
66
+
}
67
+
}
+57
src/Commands/InstallCommand.php
+57
src/Commands/InstallCommand.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Commands;
4
+
5
+
use Illuminate\Console\Command;
6
+
7
+
class InstallCommand extends Command
8
+
{
9
+
protected $signature = 'signal:install';
10
+
11
+
protected $description = 'Install the Signal package (publish config, migrations, and run migrations)';
12
+
13
+
public function handle(): int
14
+
{
15
+
$this->info('Installing Signal package...');
16
+
$this->newLine();
17
+
18
+
// Publish config
19
+
$this->comment('Publishing configuration...');
20
+
$this->callSilently('vendor:publish', [
21
+
'--tag' => 'signal-config',
22
+
'--force' => $this->option('force', false),
23
+
]);
24
+
$this->info('✓ Configuration published');
25
+
26
+
// Publish migrations
27
+
$this->comment('Publishing migrations...');
28
+
$this->callSilently('vendor:publish', [
29
+
'--tag' => 'signal-migrations',
30
+
'--force' => $this->option('force', false),
31
+
]);
32
+
$this->info('✓ Migrations published');
33
+
34
+
// Run migrations
35
+
$this->newLine();
36
+
$this->comment('Running migrations...');
37
+
38
+
if ($this->confirm('Do you want to run the migrations now?', true)) {
39
+
$this->call('migrate');
40
+
$this->info('✓ Migrations completed');
41
+
} else {
42
+
$this->warn('⚠ Skipped migrations. Run "php artisan migrate" manually when ready.');
43
+
}
44
+
45
+
$this->newLine();
46
+
$this->info('Signal package installed successfully!');
47
+
$this->newLine();
48
+
49
+
// Show next steps
50
+
$this->line('Next steps:');
51
+
$this->line('1. Review the config file: config/signal.php');
52
+
$this->line('2. Create your first signal: php artisan make:signal NewPostSignal');
53
+
$this->line('3. Start consuming events: php artisan signal:consume');
54
+
55
+
return self::SUCCESS;
56
+
}
57
+
}
+44
src/Commands/ListSignalsCommand.php
+44
src/Commands/ListSignalsCommand.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Commands;
4
+
5
+
use Illuminate\Console\Command;
6
+
use SocialDept\Signal\Services\SignalRegistry;
7
+
8
+
class ListSignalsCommand extends Command
9
+
{
10
+
protected $signature = 'signal:list';
11
+
12
+
protected $description = 'List all registered Signals';
13
+
14
+
public function handle(SignalRegistry $registry): int
15
+
{
16
+
$registry->discover();
17
+
18
+
$signals = $registry->all();
19
+
20
+
if ($signals->isEmpty()) {
21
+
$this->warn('No signals registered.');
22
+
$this->info('Create signals in app/Signals or register them in config/signal.php');
23
+
return self::SUCCESS;
24
+
}
25
+
26
+
$this->info("Found {$signals->count()} signal(s):");
27
+
$this->newLine();
28
+
29
+
$this->table(
30
+
['Signal', 'Event Types', 'Collections', 'DIDs', 'Queued'],
31
+
$signals->map(function ($signal) {
32
+
return [
33
+
get_class($signal),
34
+
implode(', ', $signal->eventTypes()),
35
+
$signal->collections() ? implode(', ', $signal->collections()) : 'All',
36
+
$signal->dids() ? implode(', ', $signal->dids()) : 'All',
37
+
$signal->shouldQueue() ? 'Yes' : 'No',
38
+
];
39
+
})
40
+
);
41
+
42
+
return self::SUCCESS;
43
+
}
44
+
}
+46
src/Commands/MakeSignalCommand.php
+46
src/Commands/MakeSignalCommand.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Commands;
4
+
5
+
use Illuminate\Console\GeneratorCommand;
6
+
use Symfony\Component\Console\Input\InputOption;
7
+
8
+
class MakeSignalCommand extends GeneratorCommand
9
+
{
10
+
protected $name = 'make:signal';
11
+
12
+
protected $description = 'Create a new Signal class';
13
+
14
+
protected $type = 'Signal';
15
+
16
+
protected function getStub(): string
17
+
{
18
+
return __DIR__ . '/../../stubs/signal.stub';
19
+
}
20
+
21
+
protected function getDefaultNamespace($rootNamespace): string
22
+
{
23
+
return $rootNamespace . '\\Signals';
24
+
}
25
+
26
+
protected function buildClass($name): string
27
+
{
28
+
$stub = parent::buildClass($name);
29
+
30
+
$eventType = $this->option('type') ?? 'commit';
31
+
$collection = $this->option('collection') ?? 'app.bsky.feed.post';
32
+
33
+
$stub = str_replace('{{ eventType }}', $eventType, $stub);
34
+
$stub = str_replace('{{ collection }}', $collection, $stub);
35
+
36
+
return $stub;
37
+
}
38
+
39
+
protected function getOptions(): array
40
+
{
41
+
return [
42
+
['type', 't', InputOption::VALUE_OPTIONAL, 'The event type (commit, identity, account)'],
43
+
['collection', 'c', InputOption::VALUE_OPTIONAL, 'The collection to watch'],
44
+
];
45
+
}
46
+
}
+84
src/Commands/TestSignalCommand.php
+84
src/Commands/TestSignalCommand.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Commands;
4
+
5
+
use Illuminate\Console\Command;
6
+
use SocialDept\Signal\Events\JetstreamEvent;
7
+
use SocialDept\Signal\Events\CommitEvent;
8
+
9
+
class TestSignalCommand extends Command
10
+
{
11
+
protected $signature = 'signal:test
12
+
{signal : The Signal class name}
13
+
{--sample=commit : The type of sample event to use}';
14
+
15
+
protected $description = 'Test a Signal with sample data';
16
+
17
+
public function handle(): int
18
+
{
19
+
$signalClass = $this->argument('signal');
20
+
21
+
// Try to resolve the class
22
+
if (!class_exists($signalClass)) {
23
+
$signalClass = 'App\\Signals\\' . $signalClass;
24
+
}
25
+
26
+
if (!class_exists($signalClass)) {
27
+
$this->error("Signal class not found: {$signalClass}");
28
+
return self::FAILURE;
29
+
}
30
+
31
+
$signal = app($signalClass);
32
+
33
+
$this->info("Testing signal: {$signalClass}");
34
+
$this->newLine();
35
+
36
+
// Create sample event
37
+
$event = $this->createSampleEvent($this->option('sample'));
38
+
39
+
$this->info('Sample event created:');
40
+
$this->line(json_encode($event->toArray(), JSON_PRETTY_PRINT));
41
+
$this->newLine();
42
+
43
+
try {
44
+
if ($signal->shouldHandle($event)) {
45
+
$this->info('Calling signal->handle()...');
46
+
$signal->handle($event);
47
+
$this->info('✓ Signal executed successfully');
48
+
} else {
49
+
$this->warn('Signal->shouldHandle() returned false');
50
+
}
51
+
} catch (\Exception $e) {
52
+
$this->error('Error executing signal: ' . $e->getMessage());
53
+
return self::FAILURE;
54
+
}
55
+
56
+
return self::SUCCESS;
57
+
}
58
+
59
+
protected function createSampleEvent(string $type): JetstreamEvent
60
+
{
61
+
switch ($type) {
62
+
case 'commit':
63
+
return new JetstreamEvent(
64
+
did: 'did:plc:sample123456789',
65
+
timeUs: time() * 1000000,
66
+
kind: 'commit',
67
+
commit: new CommitEvent(
68
+
rev: 'sample-rev',
69
+
operation: 'create',
70
+
collection: 'app.bsky.feed.post',
71
+
rkey: 'sample-rkey',
72
+
record: (object) [
73
+
'text' => 'This is a sample post for testing',
74
+
'createdAt' => now()->toIso8601String(),
75
+
],
76
+
cid: 'sample-cid',
77
+
),
78
+
);
79
+
80
+
default:
81
+
throw new \InvalidArgumentException("Unknown sample type: {$type}");
82
+
}
83
+
}
84
+
}
+21
src/Contracts/CursorStore.php
+21
src/Contracts/CursorStore.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Contracts;
4
+
5
+
interface CursorStore
6
+
{
7
+
/**
8
+
* Get the current cursor position.
9
+
*/
10
+
public function get(): ?int;
11
+
12
+
/**
13
+
* Set the cursor position.
14
+
*/
15
+
public function set(int $cursor): void;
16
+
17
+
/**
18
+
* Clear the cursor position.
19
+
*/
20
+
public function clear(): void;
21
+
}
+36
src/Events/AccountEvent.php
+36
src/Events/AccountEvent.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Events;
4
+
5
+
class AccountEvent
6
+
{
7
+
public function __construct(
8
+
public string $did,
9
+
public bool $active,
10
+
public ?string $status = null,
11
+
public int $seq = 0,
12
+
public ?string $time = null,
13
+
) {}
14
+
15
+
public static function fromArray(array $data): self
16
+
{
17
+
return new self(
18
+
did: $data['did'],
19
+
active: $data['active'],
20
+
status: $data['status'] ?? null,
21
+
seq: $data['seq'] ?? 0,
22
+
time: $data['time'] ?? null,
23
+
);
24
+
}
25
+
26
+
public function toArray(): array
27
+
{
28
+
return [
29
+
'did' => $this->did,
30
+
'active' => $this->active,
31
+
'status' => $this->status,
32
+
'seq' => $this->seq,
33
+
'time' => $this->time,
34
+
];
35
+
}
36
+
}
+59
src/Events/CommitEvent.php
+59
src/Events/CommitEvent.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Events;
4
+
5
+
class CommitEvent
6
+
{
7
+
public function __construct(
8
+
public string $rev,
9
+
public string $operation, // 'create', 'update', 'delete'
10
+
public string $collection,
11
+
public string $rkey,
12
+
public ?object $record = null,
13
+
public ?string $cid = null,
14
+
) {}
15
+
16
+
public function isCreate(): bool
17
+
{
18
+
return $this->operation === 'create';
19
+
}
20
+
21
+
public function isUpdate(): bool
22
+
{
23
+
return $this->operation === 'update';
24
+
}
25
+
26
+
public function isDelete(): bool
27
+
{
28
+
return $this->operation === 'delete';
29
+
}
30
+
31
+
public function uri(): string
32
+
{
33
+
return "at://{$this->collection}/{$this->rkey}";
34
+
}
35
+
36
+
public static function fromArray(array $data): self
37
+
{
38
+
return new self(
39
+
rev: $data['rev'],
40
+
operation: $data['operation'],
41
+
collection: $data['collection'],
42
+
rkey: $data['rkey'],
43
+
record: isset($data['record']) ? (object) $data['record'] : null,
44
+
cid: $data['cid'] ?? null,
45
+
);
46
+
}
47
+
48
+
public function toArray(): array
49
+
{
50
+
return [
51
+
'rev' => $this->rev,
52
+
'operation' => $this->operation,
53
+
'collection' => $this->collection,
54
+
'rkey' => $this->rkey,
55
+
'record' => $this->record,
56
+
'cid' => $this->cid,
57
+
];
58
+
}
59
+
}
+33
src/Events/IdentityEvent.php
+33
src/Events/IdentityEvent.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Events;
4
+
5
+
class IdentityEvent
6
+
{
7
+
public function __construct(
8
+
public string $did,
9
+
public ?string $handle = null,
10
+
public int $seq = 0,
11
+
public ?string $time = null,
12
+
) {}
13
+
14
+
public static function fromArray(array $data): self
15
+
{
16
+
return new self(
17
+
did: $data['did'],
18
+
handle: $data['handle'] ?? null,
19
+
seq: $data['seq'] ?? 0,
20
+
time: $data['time'] ?? null,
21
+
);
22
+
}
23
+
24
+
public function toArray(): array
25
+
{
26
+
return [
27
+
'did' => $this->did,
28
+
'handle' => $this->handle,
29
+
'seq' => $this->seq,
30
+
'time' => $this->time,
31
+
];
32
+
}
33
+
}
+81
src/Events/JetstreamEvent.php
+81
src/Events/JetstreamEvent.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Events;
4
+
5
+
class JetstreamEvent
6
+
{
7
+
public function __construct(
8
+
public string $did,
9
+
public int $timeUs,
10
+
public string $kind, // 'commit', 'identity', 'account'
11
+
public ?CommitEvent $commit = null,
12
+
public ?IdentityEvent $identity = null,
13
+
public ?AccountEvent $account = null,
14
+
) {}
15
+
16
+
public function isCommit(): bool
17
+
{
18
+
return $this->kind === 'commit';
19
+
}
20
+
21
+
public function isIdentity(): bool
22
+
{
23
+
return $this->kind === 'identity';
24
+
}
25
+
26
+
public function isAccount(): bool
27
+
{
28
+
return $this->kind === 'account';
29
+
}
30
+
31
+
public function getCollection(): ?string
32
+
{
33
+
return $this->commit?->collection;
34
+
}
35
+
36
+
public function getRecord(): ?object
37
+
{
38
+
return $this->commit?->record;
39
+
}
40
+
41
+
public function getOperation(): ?string
42
+
{
43
+
return $this->commit?->operation;
44
+
}
45
+
46
+
public static function fromArray(array $data): self
47
+
{
48
+
$commit = isset($data['commit'])
49
+
? CommitEvent::fromArray($data['commit'])
50
+
: null;
51
+
52
+
$identity = isset($data['identity'])
53
+
? IdentityEvent::fromArray($data['identity'])
54
+
: null;
55
+
56
+
$account = isset($data['account'])
57
+
? AccountEvent::fromArray($data['account'])
58
+
: null;
59
+
60
+
return new self(
61
+
did: $data['did'],
62
+
timeUs: $data['time_us'],
63
+
kind: $data['kind'],
64
+
commit: $commit,
65
+
identity: $identity,
66
+
account: $account,
67
+
);
68
+
}
69
+
70
+
public function toArray(): array
71
+
{
72
+
return [
73
+
'did' => $this->did,
74
+
'time_us' => $this->timeUs,
75
+
'kind' => $this->kind,
76
+
'commit' => $this->commit?->toArray(),
77
+
'identity' => $this->identity?->toArray(),
78
+
'account' => $this->account?->toArray(),
79
+
];
80
+
}
81
+
}
+8
src/Exceptions/ConnectionException.php
+8
src/Exceptions/ConnectionException.php
+8
src/Exceptions/SignalException.php
+8
src/Exceptions/SignalException.php
+8
-6
src/Facades/Signal.php
+8
-6
src/Facades/Signal.php
···
3
3
namespace SocialDept\Signal\Facades;
4
4
5
5
use Illuminate\Support\Facades\Facade;
6
+
use SocialDept\Signal\Services\JetstreamConsumer;
6
7
8
+
/**
9
+
* @method static void start(?int $cursor = null)
10
+
* @method static void stop()
11
+
*
12
+
* @see \SocialDept\Signal\Services\JetstreamConsumer
13
+
*/
7
14
class Signal extends Facade
8
15
{
9
-
/**
10
-
* Get the registered name of the component.
11
-
*
12
-
* @return string
13
-
*/
14
16
protected static function getFacadeAccessor(): string
15
17
{
16
-
return 'signal';
18
+
return JetstreamConsumer::class;
17
19
}
18
20
}
+31
src/Jobs/ProcessSignalJob.php
+31
src/Jobs/ProcessSignalJob.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Jobs;
4
+
5
+
use Illuminate\Bus\Queueable;
6
+
use Illuminate\Contracts\Queue\ShouldQueue;
7
+
use Illuminate\Foundation\Bus\Dispatchable;
8
+
use Illuminate\Queue\InteractsWithQueue;
9
+
use Illuminate\Queue\SerializesModels;
10
+
use SocialDept\Signal\Events\JetstreamEvent;
11
+
use SocialDept\Signal\Signals\Signal;
12
+
13
+
class ProcessSignalJob implements ShouldQueue
14
+
{
15
+
use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
16
+
17
+
public function __construct(
18
+
protected Signal $signal,
19
+
protected JetstreamEvent $event,
20
+
) {}
21
+
22
+
public function handle(): void
23
+
{
24
+
$this->signal->handle($this->event);
25
+
}
26
+
27
+
public function failed(\Throwable $exception): void
28
+
{
29
+
$this->signal->failed($this->event, $exception);
30
+
}
31
+
}
+61
src/Services/EventDispatcher.php
+61
src/Services/EventDispatcher.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Services;
4
+
5
+
use Illuminate\Support\Facades\Log;
6
+
use Illuminate\Support\Facades\Queue;
7
+
use SocialDept\Signal\Events\JetstreamEvent;
8
+
use SocialDept\Signal\Jobs\ProcessSignalJob;
9
+
10
+
class EventDispatcher
11
+
{
12
+
protected SignalRegistry $signalRegistry;
13
+
14
+
public function __construct(SignalRegistry $signalRegistry)
15
+
{
16
+
$this->signalRegistry = $signalRegistry;
17
+
}
18
+
19
+
/**
20
+
* Dispatch event to matching signals.
21
+
*/
22
+
public function dispatch(JetstreamEvent $event): void
23
+
{
24
+
$signals = $this->signalRegistry->getMatchingSignals($event);
25
+
26
+
foreach ($signals as $signal) {
27
+
try {
28
+
if ($signal->shouldQueue()) {
29
+
$this->dispatchToQueue($signal, $event);
30
+
} else {
31
+
$this->dispatchSync($signal, $event);
32
+
}
33
+
} catch (\Exception $e) {
34
+
Log::error('Signal: Error dispatching to signal', [
35
+
'signal' => get_class($signal),
36
+
'error' => $e->getMessage(),
37
+
]);
38
+
39
+
$signal->failed($event, $e);
40
+
}
41
+
}
42
+
}
43
+
44
+
/**
45
+
* Dispatch signal synchronously.
46
+
*/
47
+
protected function dispatchSync($signal, JetstreamEvent $event): void
48
+
{
49
+
$signal->handle($event);
50
+
}
51
+
52
+
/**
53
+
* Dispatch signal to queue.
54
+
*/
55
+
protected function dispatchToQueue($signal, JetstreamEvent $event): void
56
+
{
57
+
ProcessSignalJob::dispatch($signal, $event)
58
+
->onConnection($signal->queueConnection())
59
+
->onQueue($signal->queue());
60
+
}
61
+
}
+206
src/Services/JetstreamConsumer.php
+206
src/Services/JetstreamConsumer.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Services;
4
+
5
+
use Illuminate\Support\Facades\Log;
6
+
use Ratchet\Client\Connector;
7
+
use Ratchet\Client\WebSocket;
8
+
use React\EventLoop\Loop;
9
+
use SocialDept\Signal\Contracts\CursorStore;
10
+
use SocialDept\Signal\Events\JetstreamEvent;
11
+
use SocialDept\Signal\Exceptions\ConnectionException;
12
+
13
+
class JetstreamConsumer
14
+
{
15
+
protected CursorStore $cursorStore;
16
+
protected SignalRegistry $signalRegistry;
17
+
protected EventDispatcher $eventDispatcher;
18
+
protected ?WebSocket $connection = null;
19
+
protected int $reconnectAttempts = 0;
20
+
protected bool $shouldStop = false;
21
+
22
+
public function __construct(
23
+
CursorStore $cursorStore,
24
+
SignalRegistry $signalRegistry,
25
+
EventDispatcher $eventDispatcher
26
+
) {
27
+
$this->cursorStore = $cursorStore;
28
+
$this->signalRegistry = $signalRegistry;
29
+
$this->eventDispatcher = $eventDispatcher;
30
+
}
31
+
32
+
/**
33
+
* Start consuming the Jetstream.
34
+
*/
35
+
public function start(?int $cursor = null): void
36
+
{
37
+
$this->shouldStop = false;
38
+
39
+
// Get cursor from storage if not provided
40
+
if ($cursor === null) {
41
+
$cursor = $this->cursorStore->get();
42
+
}
43
+
44
+
$url = $this->buildWebSocketUrl($cursor);
45
+
46
+
Log::info('Signal: Starting Jetstream consumer', [
47
+
'url' => $url,
48
+
'cursor' => $cursor,
49
+
]);
50
+
51
+
$this->connect($url);
52
+
}
53
+
54
+
/**
55
+
* Stop consuming the Jetstream.
56
+
*/
57
+
public function stop(): void
58
+
{
59
+
$this->shouldStop = true;
60
+
61
+
if ($this->connection) {
62
+
$this->connection->close();
63
+
}
64
+
65
+
Log::info('Signal: Jetstream consumer stopped');
66
+
}
67
+
68
+
/**
69
+
* Connect to the Jetstream WebSocket.
70
+
*/
71
+
protected function connect(string $url): void
72
+
{
73
+
$loop = Loop::get();
74
+
$connector = new Connector($loop);
75
+
76
+
$connector($url)->then(
77
+
function (WebSocket $conn) {
78
+
$this->connection = $conn;
79
+
$this->reconnectAttempts = 0;
80
+
81
+
Log::info('Signal: Connected to Jetstream');
82
+
83
+
$conn->on('message', function ($msg) {
84
+
$this->handleMessage($msg);
85
+
});
86
+
87
+
$conn->on('close', function ($code, $reason) {
88
+
Log::warning('Signal: Connection closed', [
89
+
'code' => $code,
90
+
'reason' => $reason,
91
+
]);
92
+
93
+
if (!$this->shouldStop) {
94
+
$this->attemptReconnect();
95
+
}
96
+
});
97
+
98
+
$conn->on('error', function (\Exception $e) {
99
+
Log::error('Signal: Connection error', [
100
+
'error' => $e->getMessage(),
101
+
]);
102
+
});
103
+
104
+
// Setup ping interval to keep connection alive
105
+
$this->setupPingInterval($conn, $loop);
106
+
},
107
+
function (\Exception $e) {
108
+
Log::error('Signal: Could not connect to Jetstream', [
109
+
'error' => $e->getMessage(),
110
+
]);
111
+
112
+
if (!$this->shouldStop) {
113
+
$this->attemptReconnect();
114
+
}
115
+
}
116
+
);
117
+
118
+
$loop->run();
119
+
}
120
+
121
+
/**
122
+
* Handle incoming WebSocket message.
123
+
*/
124
+
protected function handleMessage($message): void
125
+
{
126
+
try {
127
+
$data = json_decode($message, true);
128
+
129
+
if (!$data) {
130
+
Log::warning('Signal: Failed to decode message');
131
+
return;
132
+
}
133
+
134
+
$event = JetstreamEvent::fromArray($data);
135
+
136
+
// Update cursor
137
+
$this->cursorStore->set($event->timeUs);
138
+
139
+
// Dispatch to matching signals
140
+
$this->eventDispatcher->dispatch($event);
141
+
142
+
} catch (\Exception $e) {
143
+
Log::error('Signal: Error handling message', [
144
+
'error' => $e->getMessage(),
145
+
'trace' => $e->getTraceAsString(),
146
+
]);
147
+
}
148
+
}
149
+
150
+
/**
151
+
* Attempt to reconnect to the Jetstream.
152
+
*/
153
+
protected function attemptReconnect(): void
154
+
{
155
+
$maxAttempts = config('signal.connection.reconnect_attempts', 5);
156
+
$delay = config('signal.connection.reconnect_delay', 5);
157
+
158
+
if ($this->reconnectAttempts >= $maxAttempts) {
159
+
Log::error('Signal: Max reconnection attempts reached');
160
+
throw new ConnectionException('Failed to reconnect to Jetstream after ' . $maxAttempts . ' attempts');
161
+
}
162
+
163
+
$this->reconnectAttempts++;
164
+
165
+
Log::info('Signal: Attempting to reconnect', [
166
+
'attempt' => $this->reconnectAttempts,
167
+
'max_attempts' => $maxAttempts,
168
+
]);
169
+
170
+
sleep($delay);
171
+
172
+
$cursor = $this->cursorStore->get();
173
+
$url = $this->buildWebSocketUrl($cursor);
174
+
175
+
$this->connect($url);
176
+
}
177
+
178
+
/**
179
+
* Setup ping interval to keep connection alive.
180
+
*/
181
+
protected function setupPingInterval(WebSocket $conn, $loop): void
182
+
{
183
+
$interval = config('signal.connection.ping_interval', 30);
184
+
185
+
$loop->addPeriodicTimer($interval, function () use ($conn) {
186
+
if ($conn->getReadyState() === WebSocket::STATE_OPEN) {
187
+
$conn->send(json_encode(['type' => 'ping']));
188
+
}
189
+
});
190
+
}
191
+
192
+
/**
193
+
* Build the WebSocket URL with optional cursor.
194
+
*/
195
+
protected function buildWebSocketUrl(?int $cursor = null): string
196
+
{
197
+
$baseUrl = config('signal.websocket_url', 'wss://jetstream2.us-east.bsky.network');
198
+
$url = rtrim($baseUrl, '/') . '/subscribe';
199
+
200
+
if ($cursor !== null) {
201
+
$url .= '?cursor=' . $cursor;
202
+
}
203
+
204
+
return $url;
205
+
}
206
+
}
+137
src/Services/SignalRegistry.php
+137
src/Services/SignalRegistry.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Services;
4
+
5
+
use Illuminate\Support\Collection;
6
+
use Illuminate\Support\Facades\File;
7
+
use ReflectionClass;
8
+
use SocialDept\Signal\Signals\Signal;
9
+
10
+
class SignalRegistry
11
+
{
12
+
protected Collection $signals;
13
+
14
+
public function __construct()
15
+
{
16
+
$this->signals = collect();
17
+
}
18
+
19
+
/**
20
+
* Register a signal.
21
+
*/
22
+
public function register(string $signalClass): void
23
+
{
24
+
if (!is_subclass_of($signalClass, Signal::class)) {
25
+
throw new \InvalidArgumentException(
26
+
"Signal class must extend " . Signal::class
27
+
);
28
+
}
29
+
30
+
$this->signals->push($signalClass);
31
+
}
32
+
33
+
/**
34
+
* Get all registered signals.
35
+
*/
36
+
public function all(): Collection
37
+
{
38
+
return $this->signals->map(fn($class) => app($class));
39
+
}
40
+
41
+
/**
42
+
* Auto-discover signals in the configured directory.
43
+
*/
44
+
public function discover(): void
45
+
{
46
+
if (!config('signal.auto_discovery.enabled', true)) {
47
+
return;
48
+
}
49
+
50
+
$path = config('signal.auto_discovery.path', app_path('Signals'));
51
+
$namespace = config('signal.auto_discovery.namespace', 'App\\Signals');
52
+
53
+
if (!File::exists($path)) {
54
+
return;
55
+
}
56
+
57
+
$files = File::allFiles($path);
58
+
59
+
foreach ($files as $file) {
60
+
$class = $namespace . '\\' . $file->getFilenameWithoutExtension();
61
+
62
+
if (class_exists($class) && is_subclass_of($class, Signal::class)) {
63
+
$this->register($class);
64
+
}
65
+
}
66
+
}
67
+
68
+
/**
69
+
* Get signals that match the given event.
70
+
*/
71
+
public function getMatchingSignals($event): Collection
72
+
{
73
+
return $this->all()->filter(function (Signal $signal) use ($event) {
74
+
// Check event type
75
+
if (!in_array($event->kind, $signal->eventTypes())) {
76
+
return false;
77
+
}
78
+
79
+
// Check collections filter (with wildcard support)
80
+
if ($signal->collections() !== null && $event->isCommit()) {
81
+
if (!$this->matchesCollection($event->getCollection(), $signal->collections())) {
82
+
return false;
83
+
}
84
+
}
85
+
86
+
// Check DIDs filter
87
+
if ($signal->dids() !== null) {
88
+
if (!in_array($event->did, $signal->dids())) {
89
+
return false;
90
+
}
91
+
}
92
+
93
+
// Check custom shouldHandle logic
94
+
if (!$signal->shouldHandle($event)) {
95
+
return false;
96
+
}
97
+
98
+
return true;
99
+
});
100
+
}
101
+
102
+
/**
103
+
* Check if a collection matches any of the patterns (supports wildcards).
104
+
*
105
+
* @param string|null $collection The actual collection from the event
106
+
* @param array $patterns Array of collection patterns (may include wildcards like 'app.bsky.feed.*')
107
+
* @return bool
108
+
*/
109
+
protected function matchesCollection(?string $collection, array $patterns): bool
110
+
{
111
+
if ($collection === null) {
112
+
return false;
113
+
}
114
+
115
+
foreach ($patterns as $pattern) {
116
+
// Exact match
117
+
if ($pattern === $collection) {
118
+
return true;
119
+
}
120
+
121
+
// Wildcard match
122
+
if (str_contains($pattern, '*')) {
123
+
// Convert wildcard pattern to regex
124
+
// Escape special regex characters except *
125
+
$regex = preg_quote($pattern, '/');
126
+
// Replace escaped \* with .* for regex wildcard
127
+
$regex = str_replace('\*', '.*', $regex);
128
+
129
+
if (preg_match('/^' . $regex . '$/', $collection)) {
130
+
return true;
131
+
}
132
+
}
133
+
}
134
+
135
+
return false;
136
+
}
137
+
}
-8
src/Signal.php
-8
src/Signal.php
+65
-60
src/SignalServiceProvider.php
+65
-60
src/SignalServiceProvider.php
···
3
3
namespace SocialDept\Signal;
4
4
5
5
use Illuminate\Support\ServiceProvider;
6
+
use SocialDept\Signal\Commands\ConsumeCommand;
7
+
use SocialDept\Signal\Commands\InstallCommand;
8
+
use SocialDept\Signal\Commands\ListSignalsCommand;
9
+
use SocialDept\Signal\Commands\MakeSignalCommand;
10
+
use SocialDept\Signal\Commands\TestSignalCommand;
11
+
use SocialDept\Signal\Contracts\CursorStore;
12
+
use SocialDept\Signal\Services\EventDispatcher;
13
+
use SocialDept\Signal\Services\JetstreamConsumer;
14
+
use SocialDept\Signal\Services\SignalRegistry;
15
+
use SocialDept\Signal\Storage\DatabaseCursorStore;
16
+
use SocialDept\Signal\Storage\FileCursorStore;
17
+
use SocialDept\Signal\Storage\RedisCursorStore;
6
18
7
19
class SignalServiceProvider extends ServiceProvider
8
20
{
9
-
/**
10
-
* Perform post-registration booting of services.
11
-
*
12
-
* @return void
13
-
*/
14
-
public function boot(): void
21
+
public function register(): void
15
22
{
16
-
// $this->loadTranslationsFrom(__DIR__.'/../resources/lang', 'social-dept');
17
-
// $this->loadViewsFrom(__DIR__.'/../resources/views', 'social-dept');
18
-
// $this->loadMigrationsFrom(__DIR__.'/../database/migrations');
19
-
// $this->loadRoutesFrom(__DIR__.'/routes.php');
23
+
$this->mergeConfigFrom(__DIR__ . '/../config/signal.php', 'signal');
24
+
25
+
// Register cursor store
26
+
$this->app->singleton(CursorStore::class, function ($app) {
27
+
return match (config('signal.cursor_storage')) {
28
+
'redis' => new RedisCursorStore(),
29
+
'file' => new FileCursorStore(),
30
+
default => new DatabaseCursorStore(),
31
+
};
32
+
});
33
+
34
+
// Register signal registry
35
+
$this->app->singleton(SignalRegistry::class, function ($app) {
36
+
$registry = new SignalRegistry();
20
37
21
-
// Publishing is only necessary when using the CLI.
22
-
if ($this->app->runningInConsole()) {
23
-
$this->bootForConsole();
24
-
}
25
-
}
38
+
// Register configured signals
39
+
foreach (config('signal.signals', []) as $signal) {
40
+
$registry->register($signal);
41
+
}
26
42
27
-
/**
28
-
* Register any package services.
29
-
*
30
-
* @return void
31
-
*/
32
-
public function register(): void
33
-
{
34
-
$this->mergeConfigFrom(__DIR__.'/../config/signal.php', 'signal');
43
+
return $registry;
44
+
});
35
45
36
-
// Register the service the package provides.
37
-
$this->app->singleton('signal', function ($app) {
38
-
return new Signal;
46
+
// Register event dispatcher
47
+
$this->app->singleton(EventDispatcher::class, function ($app) {
48
+
return new EventDispatcher($app->make(SignalRegistry::class));
39
49
});
40
-
}
41
50
42
-
/**
43
-
* Get the services provided by the provider.
44
-
*
45
-
* @return array
46
-
*/
47
-
public function provides()
48
-
{
49
-
return ['signal'];
51
+
// Register Jetstream consumer
52
+
$this->app->singleton(JetstreamConsumer::class, function ($app) {
53
+
return new JetstreamConsumer(
54
+
$app->make(CursorStore::class),
55
+
$app->make(SignalRegistry::class),
56
+
$app->make(EventDispatcher::class),
57
+
);
58
+
});
50
59
}
51
60
52
-
/**
53
-
* Console-specific booting.
54
-
*
55
-
* @return void
56
-
*/
57
-
protected function bootForConsole(): void
61
+
public function boot(): void
58
62
{
59
-
// Publishing the configuration file.
60
-
$this->publishes([
61
-
__DIR__.'/../config/signal.php' => config_path('signal.php'),
62
-
], 'signal.config');
63
-
64
-
// Publishing the views.
65
-
/*$this->publishes([
66
-
__DIR__.'/../resources/views' => base_path('resources/views/vendor/social-dept'),
67
-
], 'signal.views');*/
63
+
if ($this->app->runningInConsole()) {
64
+
// Publish config
65
+
$this->publishes([
66
+
__DIR__ . '/../config/signal.php' => config_path('signal.php'),
67
+
], 'signal-config');
68
68
69
-
// Publishing assets.
70
-
/*$this->publishes([
71
-
__DIR__.'/../resources/assets' => public_path('vendor/social-dept'),
72
-
], 'signal.assets');*/
69
+
// Publish migrations
70
+
$this->publishes([
71
+
__DIR__ . '/../database/migrations' => database_path('migrations'),
72
+
], 'signal-migrations');
73
73
74
-
// Publishing the translation files.
75
-
/*$this->publishes([
76
-
__DIR__.'/../resources/lang' => resource_path('lang/vendor/social-dept'),
77
-
], 'signal.lang');*/
74
+
// Register commands
75
+
$this->commands([
76
+
InstallCommand::class,
77
+
ConsumeCommand::class,
78
+
ListSignalsCommand::class,
79
+
MakeSignalCommand::class,
80
+
TestSignalCommand::class,
81
+
]);
82
+
}
78
83
79
-
// Registering package commands.
80
-
// $this->commands([]);
84
+
// Load migrations
85
+
$this->loadMigrationsFrom(__DIR__ . '/../database/migrations');
81
86
}
82
87
}
+97
src/Signals/Signal.php
+97
src/Signals/Signal.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Signals;
4
+
5
+
use SocialDept\Signal\Events\JetstreamEvent;
6
+
7
+
abstract class Signal
8
+
{
9
+
/**
10
+
* Define which event types to listen for.
11
+
*
12
+
* @return array<string> ['commit', 'identity', 'account']
13
+
*/
14
+
abstract public function eventTypes(): array;
15
+
16
+
/**
17
+
* Define collections to watch (optional, null = all).
18
+
* Supports wildcards using asterisk (*).
19
+
*
20
+
* Examples:
21
+
* - ['app.bsky.feed.post'] - Only posts
22
+
* - ['app.bsky.feed.*'] - All feed collections (post, like, repost, etc.)
23
+
* - ['app.bsky.graph.*'] - All graph collections (follow, block, etc.)
24
+
* - ['app.bsky.*'] - All app.bsky collections
25
+
*
26
+
* @return array<string>|null
27
+
*/
28
+
public function collections(): ?array
29
+
{
30
+
return null;
31
+
}
32
+
33
+
/**
34
+
* Define DIDs to watch (optional, null = all).
35
+
*
36
+
* @return array<string>|null
37
+
*/
38
+
public function dids(): ?array
39
+
{
40
+
return null;
41
+
}
42
+
43
+
/**
44
+
* Handle the Jetstream event.
45
+
*/
46
+
abstract public function handle(JetstreamEvent $event): void;
47
+
48
+
/**
49
+
* Determine if this signal should handle the event.
50
+
*/
51
+
public function shouldHandle(JetstreamEvent $event): bool
52
+
{
53
+
return true;
54
+
}
55
+
56
+
/**
57
+
* Should this signal be queued?
58
+
*/
59
+
public function shouldQueue(): bool
60
+
{
61
+
return false;
62
+
}
63
+
64
+
/**
65
+
* Get the queue connection name.
66
+
*/
67
+
public function queueConnection(): ?string
68
+
{
69
+
return config('signal.queue.connection');
70
+
}
71
+
72
+
/**
73
+
* Get the queue name.
74
+
*/
75
+
public function queue(): ?string
76
+
{
77
+
return config('signal.queue.queue');
78
+
}
79
+
80
+
/**
81
+
* Middleware to run before handling the event.
82
+
*
83
+
* @return array
84
+
*/
85
+
public function middleware(): array
86
+
{
87
+
return [];
88
+
}
89
+
90
+
/**
91
+
* Handle a failed signal execution.
92
+
*/
93
+
public function failed(JetstreamEvent $event, \Throwable $exception): void
94
+
{
95
+
//
96
+
}
97
+
}
+52
src/Storage/DatabaseCursorStore.php
+52
src/Storage/DatabaseCursorStore.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Storage;
4
+
5
+
use Illuminate\Support\Facades\DB;
6
+
use SocialDept\Signal\Contracts\CursorStore;
7
+
8
+
class DatabaseCursorStore implements CursorStore
9
+
{
10
+
protected string $table;
11
+
protected ?string $connection;
12
+
13
+
public function __construct()
14
+
{
15
+
$this->table = config('signal.cursor_config.database.table', 'signal_cursors');
16
+
$this->connection = config('signal.cursor_config.database.connection');
17
+
}
18
+
19
+
public function get(): ?int
20
+
{
21
+
$cursor = $this->query()
22
+
->where('key', 'jetstream')
23
+
->value('cursor');
24
+
25
+
return $cursor ? (int) $cursor : null;
26
+
}
27
+
28
+
public function set(int $cursor): void
29
+
{
30
+
$this->query()
31
+
->updateOrInsert(
32
+
['key' => 'jetstream'],
33
+
[
34
+
'cursor' => $cursor,
35
+
'updated_at' => now(),
36
+
]
37
+
);
38
+
}
39
+
40
+
public function clear(): void
41
+
{
42
+
$this->query()
43
+
->where('key', 'jetstream')
44
+
->delete();
45
+
}
46
+
47
+
protected function query()
48
+
{
49
+
return DB::connection($this->connection)
50
+
->table($this->table);
51
+
}
52
+
}
+48
src/Storage/FileCursorStore.php
+48
src/Storage/FileCursorStore.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Storage;
4
+
5
+
use Illuminate\Support\Facades\File;
6
+
use SocialDept\Signal\Contracts\CursorStore;
7
+
8
+
class FileCursorStore implements CursorStore
9
+
{
10
+
protected string $path;
11
+
12
+
public function __construct()
13
+
{
14
+
$this->path = config('signal.cursor_config.file.path', storage_path('signal/cursor.json'));
15
+
16
+
// Ensure directory exists
17
+
$directory = dirname($this->path);
18
+
if (!File::exists($directory)) {
19
+
File::makeDirectory($directory, 0755, true);
20
+
}
21
+
}
22
+
23
+
public function get(): ?int
24
+
{
25
+
if (!File::exists($this->path)) {
26
+
return null;
27
+
}
28
+
29
+
$data = json_decode(File::get($this->path), true);
30
+
31
+
return $data['cursor'] ?? null;
32
+
}
33
+
34
+
public function set(int $cursor): void
35
+
{
36
+
File::put($this->path, json_encode([
37
+
'cursor' => $cursor,
38
+
'updated_at' => now()->toIso8601String(),
39
+
], JSON_PRETTY_PRINT));
40
+
}
41
+
42
+
public function clear(): void
43
+
{
44
+
if (File::exists($this->path)) {
45
+
File::delete($this->path);
46
+
}
47
+
}
48
+
}
+35
src/Storage/RedisCursorStore.php
+35
src/Storage/RedisCursorStore.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Storage;
4
+
5
+
use Illuminate\Support\Facades\Redis;
6
+
use SocialDept\Signal\Contracts\CursorStore;
7
+
8
+
class RedisCursorStore implements CursorStore
9
+
{
10
+
protected string $connection;
11
+
protected string $key;
12
+
13
+
public function __construct()
14
+
{
15
+
$this->connection = config('signal.cursor_config.redis.connection', 'default');
16
+
$this->key = config('signal.cursor_config.redis.key', 'signal:cursor');
17
+
}
18
+
19
+
public function get(): ?int
20
+
{
21
+
$cursor = Redis::connection($this->connection)->get($this->key);
22
+
23
+
return $cursor ? (int) $cursor : null;
24
+
}
25
+
26
+
public function set(int $cursor): void
27
+
{
28
+
Redis::connection($this->connection)->set($this->key, $cursor);
29
+
}
30
+
31
+
public function clear(): void
32
+
{
33
+
Redis::connection($this->connection)->del($this->key);
34
+
}
35
+
}
+63
stubs/signal.stub
+63
stubs/signal.stub
···
1
+
<?php
2
+
3
+
namespace {{ namespace }};
4
+
5
+
use SocialDept\Signal\Events\JetstreamEvent;
6
+
use SocialDept\Signal\Signals\Signal;
7
+
8
+
class {{ class }} extends Signal
9
+
{
10
+
/**
11
+
* Define which event types to listen for.
12
+
*/
13
+
public function eventTypes(): array
14
+
{
15
+
return ['{{ eventType }}'];
16
+
}
17
+
18
+
/**
19
+
* Define collections to watch (optional, null = all).
20
+
*
21
+
* Supports wildcards:
22
+
* - ['app.bsky.feed.post'] - Only posts
23
+
* - ['app.bsky.feed.*'] - All feed collections
24
+
* - ['app.bsky.graph.*'] - All graph collections
25
+
*/
26
+
public function collections(): ?array
27
+
{
28
+
return ['{{ collection }}'];
29
+
}
30
+
31
+
/**
32
+
* Handle the Jetstream event.
33
+
*/
34
+
public function handle(JetstreamEvent $event): void
35
+
{
36
+
// Handle the event here
37
+
38
+
// Example: Access commit data
39
+
if ($event->isCommit()) {
40
+
$record = $event->getRecord();
41
+
$operation = $event->getOperation();
42
+
$collection = $event->getCollection();
43
+
44
+
// Your logic here
45
+
}
46
+
}
47
+
48
+
/**
49
+
* Determine if this signal should handle the event.
50
+
*/
51
+
public function shouldHandle(JetstreamEvent $event): bool
52
+
{
53
+
return true;
54
+
}
55
+
56
+
/**
57
+
* Should this signal be queued?
58
+
*/
59
+
public function shouldQueue(): bool
60
+
{
61
+
return false;
62
+
}
63
+
}
+136
tests/Unit/SignalRegistryTest.php
+136
tests/Unit/SignalRegistryTest.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Tests\Unit;
4
+
5
+
use Orchestra\Testbench\TestCase;
6
+
use SocialDept\Signal\Events\CommitEvent;
7
+
use SocialDept\Signal\Events\JetstreamEvent;
8
+
use SocialDept\Signal\Services\SignalRegistry;
9
+
use SocialDept\Signal\Signals\Signal;
10
+
11
+
class SignalRegistryTest extends TestCase
12
+
{
13
+
/** @test */
14
+
public function it_matches_exact_collections()
15
+
{
16
+
$registry = new SignalRegistry();
17
+
18
+
$event = new JetstreamEvent(
19
+
did: 'did:plc:test',
20
+
timeUs: time() * 1000000,
21
+
kind: 'commit',
22
+
commit: new CommitEvent(
23
+
rev: 'test',
24
+
operation: 'create',
25
+
collection: 'app.bsky.feed.post',
26
+
rkey: 'test',
27
+
),
28
+
);
29
+
30
+
$result = $this->invokeMethod(
31
+
$registry,
32
+
'matchesCollection',
33
+
['app.bsky.feed.post', ['app.bsky.feed.post']]
34
+
);
35
+
36
+
$this->assertTrue($result);
37
+
}
38
+
39
+
/** @test */
40
+
public function it_matches_wildcard_collections()
41
+
{
42
+
$registry = new SignalRegistry();
43
+
44
+
// Test app.bsky.feed.*
45
+
$this->assertTrue(
46
+
$this->invokeMethod(
47
+
$registry,
48
+
'matchesCollection',
49
+
['app.bsky.feed.post', ['app.bsky.feed.*']]
50
+
)
51
+
);
52
+
53
+
$this->assertTrue(
54
+
$this->invokeMethod(
55
+
$registry,
56
+
'matchesCollection',
57
+
['app.bsky.feed.like', ['app.bsky.feed.*']]
58
+
)
59
+
);
60
+
61
+
$this->assertFalse(
62
+
$this->invokeMethod(
63
+
$registry,
64
+
'matchesCollection',
65
+
['app.bsky.graph.follow', ['app.bsky.feed.*']]
66
+
)
67
+
);
68
+
69
+
// Test app.bsky.*
70
+
$this->assertTrue(
71
+
$this->invokeMethod(
72
+
$registry,
73
+
'matchesCollection',
74
+
['app.bsky.feed.post', ['app.bsky.*']]
75
+
)
76
+
);
77
+
78
+
$this->assertTrue(
79
+
$this->invokeMethod(
80
+
$registry,
81
+
'matchesCollection',
82
+
['app.bsky.graph.follow', ['app.bsky.*']]
83
+
)
84
+
);
85
+
}
86
+
87
+
/** @test */
88
+
public function it_matches_multiple_patterns()
89
+
{
90
+
$registry = new SignalRegistry();
91
+
92
+
$patterns = [
93
+
'app.bsky.feed.post',
94
+
'app.bsky.graph.*',
95
+
];
96
+
97
+
// Exact match
98
+
$this->assertTrue(
99
+
$this->invokeMethod(
100
+
$registry,
101
+
'matchesCollection',
102
+
['app.bsky.feed.post', $patterns]
103
+
)
104
+
);
105
+
106
+
// Wildcard match
107
+
$this->assertTrue(
108
+
$this->invokeMethod(
109
+
$registry,
110
+
'matchesCollection',
111
+
['app.bsky.graph.follow', $patterns]
112
+
)
113
+
);
114
+
115
+
// No match
116
+
$this->assertFalse(
117
+
$this->invokeMethod(
118
+
$registry,
119
+
'matchesCollection',
120
+
['app.bsky.feed.like', $patterns]
121
+
)
122
+
);
123
+
}
124
+
125
+
/**
126
+
* Call protected/private method of a class.
127
+
*/
128
+
protected function invokeMethod(&$object, $methodName, array $parameters = [])
129
+
{
130
+
$reflection = new \ReflectionClass(get_class($object));
131
+
$method = $reflection->getMethod($methodName);
132
+
$method->setAccessible(true);
133
+
134
+
return $method->invokeArgs($object, $parameters);
135
+
}
136
+
}
+131
tests/Unit/SignalTest.php
+131
tests/Unit/SignalTest.php
···
1
+
<?php
2
+
3
+
namespace SocialDept\Signal\Tests\Unit;
4
+
5
+
use Orchestra\Testbench\TestCase;
6
+
use SocialDept\Signal\Events\CommitEvent;
7
+
use SocialDept\Signal\Events\JetstreamEvent;
8
+
use SocialDept\Signal\Signals\Signal;
9
+
10
+
class SignalTest extends TestCase
11
+
{
12
+
/** @test */
13
+
public function it_can_create_a_signal()
14
+
{
15
+
$signal = new class extends Signal {
16
+
public function eventTypes(): array
17
+
{
18
+
return ['commit'];
19
+
}
20
+
21
+
public function handle(JetstreamEvent $event): void
22
+
{
23
+
//
24
+
}
25
+
};
26
+
27
+
$this->assertInstanceOf(Signal::class, $signal);
28
+
$this->assertEquals(['commit'], $signal->eventTypes());
29
+
}
30
+
31
+
/** @test */
32
+
public function it_can_filter_by_exact_collection()
33
+
{
34
+
$signal = new class extends Signal {
35
+
public function eventTypes(): array
36
+
{
37
+
return ['commit'];
38
+
}
39
+
40
+
public function collections(): ?array
41
+
{
42
+
return ['app.bsky.feed.post'];
43
+
}
44
+
45
+
public function handle(JetstreamEvent $event): void
46
+
{
47
+
//
48
+
}
49
+
};
50
+
51
+
$event = new JetstreamEvent(
52
+
did: 'did:plc:test',
53
+
timeUs: time() * 1000000,
54
+
kind: 'commit',
55
+
commit: new CommitEvent(
56
+
rev: 'test',
57
+
operation: 'create',
58
+
collection: 'app.bsky.feed.post',
59
+
rkey: 'test',
60
+
),
61
+
);
62
+
63
+
$this->assertTrue($signal->shouldHandle($event));
64
+
}
65
+
66
+
/** @test */
67
+
public function it_can_filter_by_wildcard_collection()
68
+
{
69
+
$signal = new class extends Signal {
70
+
public function eventTypes(): array
71
+
{
72
+
return ['commit'];
73
+
}
74
+
75
+
public function collections(): ?array
76
+
{
77
+
return ['app.bsky.feed.*'];
78
+
}
79
+
80
+
public function handle(JetstreamEvent $event): void
81
+
{
82
+
//
83
+
}
84
+
};
85
+
86
+
// Test that it matches app.bsky.feed.post
87
+
$postEvent = new JetstreamEvent(
88
+
did: 'did:plc:test',
89
+
timeUs: time() * 1000000,
90
+
kind: 'commit',
91
+
commit: new CommitEvent(
92
+
rev: 'test',
93
+
operation: 'create',
94
+
collection: 'app.bsky.feed.post',
95
+
rkey: 'test',
96
+
),
97
+
);
98
+
99
+
$this->assertTrue($signal->shouldHandle($postEvent));
100
+
101
+
// Test that it matches app.bsky.feed.like
102
+
$likeEvent = new JetstreamEvent(
103
+
did: 'did:plc:test',
104
+
timeUs: time() * 1000000,
105
+
kind: 'commit',
106
+
commit: new CommitEvent(
107
+
rev: 'test',
108
+
operation: 'create',
109
+
collection: 'app.bsky.feed.like',
110
+
rkey: 'test',
111
+
),
112
+
);
113
+
114
+
$this->assertTrue($signal->shouldHandle($likeEvent));
115
+
116
+
// Test that it does NOT match app.bsky.graph.follow
117
+
$followEvent = new JetstreamEvent(
118
+
did: 'did:plc:test',
119
+
timeUs: time() * 1000000,
120
+
kind: 'commit',
121
+
commit: new CommitEvent(
122
+
rev: 'test',
123
+
operation: 'create',
124
+
collection: 'app.bsky.graph.follow',
125
+
rkey: 'test',
126
+
),
127
+
);
128
+
129
+
$this->assertFalse($signal->shouldHandle($followEvent));
130
+
}
131
+
}