Maintain local ⭤ remote in sync with automatic AT Protocol parity for Laravel (alpha & unstable)
Add documentation
+2286
+343
docs/atp-client-integration.md
+343
docs/atp-client-integration.md
···
1
+
# atp-client Integration
2
+
3
+
Parity integrates with atp-client to fetch records from the AT Protocol network and convert them to Eloquent models. The `RecordHelper` class provides a simple interface for these operations.
4
+
5
+
## RecordHelper
6
+
7
+
The `RecordHelper` is registered as a singleton and available via the container:
8
+
9
+
```php
10
+
use SocialDept\AtpParity\Support\RecordHelper;
11
+
12
+
$helper = app(RecordHelper::class);
13
+
```
14
+
15
+
### How It Works
16
+
17
+
When you provide an AT Protocol URI, RecordHelper:
18
+
19
+
1. Parses the URI to extract the DID, collection, and rkey
20
+
2. Resolves the DID to find the user's PDS endpoint (via atp-resolver)
21
+
3. Creates a public client for that PDS
22
+
4. Fetches the record
23
+
5. Converts it using the registered mapper
24
+
25
+
This means it works with any AT Protocol server, not just Bluesky.
26
+
27
+
## Fetching Records
28
+
29
+
### `fetch(string $uri, ?string $recordClass = null): mixed`
30
+
31
+
Fetches a record and returns it as a typed DTO.
32
+
33
+
```php
34
+
use SocialDept\AtpParity\Support\RecordHelper;
35
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
36
+
37
+
$helper = app(RecordHelper::class);
38
+
39
+
// Auto-detect type from registered mapper
40
+
$record = $helper->fetch('at://did:plc:abc123/app.bsky.feed.post/xyz789');
41
+
42
+
// Or specify the class explicitly
43
+
$record = $helper->fetch(
44
+
'at://did:plc:abc123/app.bsky.feed.post/xyz789',
45
+
Post::class
46
+
);
47
+
48
+
// Access typed properties
49
+
echo $record->text;
50
+
echo $record->createdAt;
51
+
```
52
+
53
+
### `fetchAsModel(string $uri): ?Model`
54
+
55
+
Fetches a record and converts it to an Eloquent model (unsaved).
56
+
57
+
```php
58
+
$post = $helper->fetchAsModel('at://did:plc:abc123/app.bsky.feed.post/xyz789');
59
+
60
+
if ($post) {
61
+
echo $post->content;
62
+
echo $post->atp_uri;
63
+
echo $post->atp_cid;
64
+
65
+
// Save if you want to persist it
66
+
$post->save();
67
+
}
68
+
```
69
+
70
+
Returns `null` if no mapper is registered for the collection.
71
+
72
+
### `sync(string $uri): ?Model`
73
+
74
+
Fetches a record and upserts it to the database.
75
+
76
+
```php
77
+
// Creates or updates the model
78
+
$post = $helper->sync('at://did:plc:abc123/app.bsky.feed.post/xyz789');
79
+
80
+
// Model is saved automatically
81
+
echo $post->id;
82
+
echo $post->content;
83
+
```
84
+
85
+
This is the most common method for syncing remote records to your database.
86
+
87
+
## Working with Responses
88
+
89
+
### `hydrateRecord(GetRecordResponse $response, ?string $recordClass = null): mixed`
90
+
91
+
If you already have a `GetRecordResponse` from atp-client, convert it to a typed DTO:
92
+
93
+
```php
94
+
use SocialDept\AtpClient\Facades\Atp;
95
+
use SocialDept\AtpParity\Support\RecordHelper;
96
+
97
+
$helper = app(RecordHelper::class);
98
+
99
+
// Using atp-client directly
100
+
$client = Atp::public();
101
+
$response = $client->atproto->repo->getRecord(
102
+
'did:plc:abc123',
103
+
'app.bsky.feed.post',
104
+
'xyz789'
105
+
);
106
+
107
+
// Convert to typed DTO
108
+
$record = $helper->hydrateRecord($response);
109
+
```
110
+
111
+
## Practical Examples
112
+
113
+
### Syncing a Single Post
114
+
115
+
```php
116
+
$helper = app(RecordHelper::class);
117
+
118
+
$uri = 'at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.post/3k2yihcrp6f2c';
119
+
$post = $helper->sync($uri);
120
+
121
+
echo "Synced: {$post->content}";
122
+
```
123
+
124
+
### Syncing Multiple Posts
125
+
126
+
```php
127
+
$helper = app(RecordHelper::class);
128
+
129
+
$uris = [
130
+
'at://did:plc:abc/app.bsky.feed.post/123',
131
+
'at://did:plc:def/app.bsky.feed.post/456',
132
+
'at://did:plc:ghi/app.bsky.feed.post/789',
133
+
];
134
+
135
+
foreach ($uris as $uri) {
136
+
try {
137
+
$post = $helper->sync($uri);
138
+
echo "Synced: {$post->id}\n";
139
+
} catch (\Exception $e) {
140
+
echo "Failed to sync {$uri}: {$e->getMessage()}\n";
141
+
}
142
+
}
143
+
```
144
+
145
+
### Fetching for Preview (Without Saving)
146
+
147
+
```php
148
+
$helper = app(RecordHelper::class);
149
+
150
+
// Get model without saving
151
+
$post = $helper->fetchAsModel('at://did:plc:xxx/app.bsky.feed.post/abc');
152
+
153
+
if ($post) {
154
+
return view('posts.preview', ['post' => $post]);
155
+
}
156
+
157
+
return abort(404);
158
+
```
159
+
160
+
### Checking if Record Exists Locally
161
+
162
+
```php
163
+
use App\Models\Post;
164
+
use SocialDept\AtpParity\Support\RecordHelper;
165
+
166
+
$uri = 'at://did:plc:xxx/app.bsky.feed.post/abc';
167
+
168
+
// Check local database first
169
+
$post = Post::whereAtpUri($uri)->first();
170
+
171
+
if (!$post) {
172
+
// Not in database, fetch from network
173
+
$helper = app(RecordHelper::class);
174
+
$post = $helper->sync($uri);
175
+
}
176
+
177
+
return $post;
178
+
```
179
+
180
+
### Building a Post Importer
181
+
182
+
```php
183
+
namespace App\Services;
184
+
185
+
use SocialDept\AtpParity\Support\RecordHelper;
186
+
use SocialDept\AtpClient\Facades\Atp;
187
+
188
+
class PostImporter
189
+
{
190
+
public function __construct(
191
+
protected RecordHelper $helper
192
+
) {}
193
+
194
+
/**
195
+
* Import all posts from a user.
196
+
*/
197
+
public function importUserPosts(string $did, int $limit = 100): array
198
+
{
199
+
$imported = [];
200
+
$client = Atp::public();
201
+
$cursor = null;
202
+
203
+
do {
204
+
$response = $client->atproto->repo->listRecords(
205
+
repo: $did,
206
+
collection: 'app.bsky.feed.post',
207
+
limit: min($limit - count($imported), 100),
208
+
cursor: $cursor
209
+
);
210
+
211
+
foreach ($response->records as $record) {
212
+
$post = $this->helper->sync($record->uri);
213
+
$imported[] = $post;
214
+
215
+
if (count($imported) >= $limit) {
216
+
break 2;
217
+
}
218
+
}
219
+
220
+
$cursor = $response->cursor;
221
+
} while ($cursor && count($imported) < $limit);
222
+
223
+
return $imported;
224
+
}
225
+
}
226
+
```
227
+
228
+
## Error Handling
229
+
230
+
RecordHelper returns `null` for various failure conditions:
231
+
232
+
```php
233
+
$helper = app(RecordHelper::class);
234
+
235
+
// Invalid URI format
236
+
$result = $helper->fetch('not-a-valid-uri');
237
+
// Returns null
238
+
239
+
// No mapper registered for collection
240
+
$result = $helper->fetchAsModel('at://did:plc:xxx/some.unknown.collection/abc');
241
+
// Returns null
242
+
243
+
// PDS resolution failed
244
+
$result = $helper->fetch('at://did:plc:invalid/app.bsky.feed.post/abc');
245
+
// Returns null (or throws exception depending on resolver config)
246
+
```
247
+
248
+
For more control, catch exceptions:
249
+
250
+
```php
251
+
use SocialDept\AtpResolver\Exceptions\DidResolutionException;
252
+
253
+
try {
254
+
$post = $helper->sync($uri);
255
+
} catch (DidResolutionException $e) {
256
+
// DID could not be resolved
257
+
Log::warning("Could not resolve DID for {$uri}");
258
+
} catch (\Exception $e) {
259
+
// Network error, invalid response, etc.
260
+
Log::error("Failed to sync {$uri}: {$e->getMessage()}");
261
+
}
262
+
```
263
+
264
+
## Performance Considerations
265
+
266
+
### PDS Client Caching
267
+
268
+
RecordHelper caches public clients by PDS endpoint:
269
+
270
+
```php
271
+
// First request to this PDS - creates client
272
+
$helper->sync('at://did:plc:abc/app.bsky.feed.post/1');
273
+
274
+
// Same PDS - reuses cached client
275
+
$helper->sync('at://did:plc:abc/app.bsky.feed.post/2');
276
+
277
+
// Different PDS - creates new client
278
+
$helper->sync('at://did:plc:xyz/app.bsky.feed.post/1');
279
+
```
280
+
281
+
### DID Resolution Caching
282
+
283
+
atp-resolver caches DID documents and PDS endpoints. Default TTL is 1 hour.
284
+
285
+
### Batch Operations
286
+
287
+
For bulk imports, consider using atp-client's `listRecords` directly and then batch-processing:
288
+
289
+
```php
290
+
use SocialDept\AtpClient\Facades\Atp;
291
+
use SocialDept\AtpParity\MapperRegistry;
292
+
293
+
$client = Atp::public($pdsEndpoint);
294
+
$registry = app(MapperRegistry::class);
295
+
$mapper = $registry->forLexicon('app.bsky.feed.post');
296
+
297
+
$response = $client->atproto->repo->listRecords(
298
+
repo: $did,
299
+
collection: 'app.bsky.feed.post',
300
+
limit: 100
301
+
);
302
+
303
+
foreach ($response->records as $record) {
304
+
$recordClass = $mapper->recordClass();
305
+
$dto = $recordClass::fromArray($record->value);
306
+
307
+
$mapper->upsert($dto, [
308
+
'uri' => $record->uri,
309
+
'cid' => $record->cid,
310
+
]);
311
+
}
312
+
```
313
+
314
+
## Using with Authenticated Client
315
+
316
+
While RecordHelper uses public clients, you can also use authenticated clients for records that require auth:
317
+
318
+
```php
319
+
use SocialDept\AtpClient\Facades\Atp;
320
+
use SocialDept\AtpParity\MapperRegistry;
321
+
322
+
// Authenticated client
323
+
$client = Atp::as('user.bsky.social');
324
+
325
+
// Fetch a record that requires auth
326
+
$response = $client->atproto->repo->getRecord(
327
+
repo: $client->session()->did(),
328
+
collection: 'app.bsky.feed.post',
329
+
rkey: 'abc123'
330
+
);
331
+
332
+
// Convert using mapper
333
+
$registry = app(MapperRegistry::class);
334
+
$mapper = $registry->forLexicon('app.bsky.feed.post');
335
+
336
+
$recordClass = $mapper->recordClass();
337
+
$record = $recordClass::fromArray($response->value);
338
+
339
+
$model = $mapper->upsert($record, [
340
+
'uri' => $response->uri,
341
+
'cid' => $response->cid,
342
+
]);
343
+
```
+355
docs/atp-schema-integration.md
+355
docs/atp-schema-integration.md
···
1
+
# atp-schema Integration
2
+
3
+
Parity is built on top of atp-schema, using its `Data` base class for all record DTOs. This provides type safety, validation, and compatibility with the AT Protocol ecosystem.
4
+
5
+
## How It Works
6
+
7
+
The `SocialDept\AtpParity\Data\Record` class extends `SocialDept\AtpSchema\Data\Data`:
8
+
9
+
```php
10
+
namespace SocialDept\AtpParity\Data;
11
+
12
+
use SocialDept\AtpClient\Contracts\Recordable;
13
+
use SocialDept\AtpSchema\Data\Data;
14
+
15
+
abstract class Record extends Data implements Recordable
16
+
{
17
+
public function getType(): string
18
+
{
19
+
return static::getLexicon();
20
+
}
21
+
}
22
+
```
23
+
24
+
This means all Parity records inherit:
25
+
26
+
- `getLexicon()` - Returns the lexicon NSID
27
+
- `fromArray()` - Creates instance from array data
28
+
- `toArray()` - Converts to array
29
+
- `toRecord()` - Converts to record format for API calls
30
+
- Type validation and casting
31
+
32
+
## Using Generated Schema Classes
33
+
34
+
atp-schema generates PHP classes for all AT Protocol lexicons. Use them directly with Parity:
35
+
36
+
```php
37
+
use SocialDept\AtpParity\Support\SchemaMapper;
38
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
39
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Like;
40
+
use SocialDept\AtpSchema\Generated\App\Bsky\Graph\Follow;
41
+
42
+
// Post mapper
43
+
$postMapper = new SchemaMapper(
44
+
schemaClass: Post::class,
45
+
modelClass: \App\Models\Post::class,
46
+
toAttributes: fn(Post $post) => [
47
+
'content' => $post->text,
48
+
'published_at' => $post->createdAt,
49
+
'langs' => $post->langs,
50
+
'reply_parent' => $post->reply?->parent->uri,
51
+
'reply_root' => $post->reply?->root->uri,
52
+
],
53
+
toRecordData: fn($model) => [
54
+
'text' => $model->content,
55
+
'createdAt' => $model->published_at->toIso8601String(),
56
+
'langs' => $model->langs ?? ['en'],
57
+
],
58
+
);
59
+
60
+
// Like mapper
61
+
$likeMapper = new SchemaMapper(
62
+
schemaClass: Like::class,
63
+
modelClass: \App\Models\Like::class,
64
+
toAttributes: fn(Like $like) => [
65
+
'subject_uri' => $like->subject->uri,
66
+
'subject_cid' => $like->subject->cid,
67
+
'liked_at' => $like->createdAt,
68
+
],
69
+
toRecordData: fn($model) => [
70
+
'subject' => [
71
+
'uri' => $model->subject_uri,
72
+
'cid' => $model->subject_cid,
73
+
],
74
+
'createdAt' => $model->liked_at->toIso8601String(),
75
+
],
76
+
);
77
+
78
+
// Follow mapper
79
+
$followMapper = new SchemaMapper(
80
+
schemaClass: Follow::class,
81
+
modelClass: \App\Models\Follow::class,
82
+
toAttributes: fn(Follow $follow) => [
83
+
'subject_did' => $follow->subject,
84
+
'followed_at' => $follow->createdAt,
85
+
],
86
+
toRecordData: fn($model) => [
87
+
'subject' => $model->subject_did,
88
+
'createdAt' => $model->followed_at->toIso8601String(),
89
+
],
90
+
);
91
+
```
92
+
93
+
## Creating Custom Records
94
+
95
+
For custom lexicons or when you need more control, extend the `Record` class:
96
+
97
+
```php
98
+
<?php
99
+
100
+
namespace App\AtpRecords;
101
+
102
+
use Carbon\Carbon;
103
+
use SocialDept\AtpParity\Data\Record;
104
+
105
+
class CustomPost extends Record
106
+
{
107
+
public function __construct(
108
+
public readonly string $text,
109
+
public readonly Carbon $createdAt,
110
+
public readonly ?array $facets = null,
111
+
public readonly ?array $embed = null,
112
+
public readonly ?array $langs = null,
113
+
) {}
114
+
115
+
public static function getLexicon(): string
116
+
{
117
+
return 'app.bsky.feed.post';
118
+
}
119
+
120
+
public static function fromArray(array $data): static
121
+
{
122
+
return new static(
123
+
text: $data['text'],
124
+
createdAt: Carbon::parse($data['createdAt']),
125
+
facets: $data['facets'] ?? null,
126
+
embed: $data['embed'] ?? null,
127
+
langs: $data['langs'] ?? null,
128
+
);
129
+
}
130
+
131
+
public function toArray(): array
132
+
{
133
+
return array_filter([
134
+
'$type' => static::getLexicon(),
135
+
'text' => $this->text,
136
+
'createdAt' => $this->createdAt->toIso8601String(),
137
+
'facets' => $this->facets,
138
+
'embed' => $this->embed,
139
+
'langs' => $this->langs,
140
+
], fn($v) => $v !== null);
141
+
}
142
+
}
143
+
```
144
+
145
+
## Custom Lexicons (AppView)
146
+
147
+
Building a custom AT Protocol application? Define your own lexicons:
148
+
149
+
```php
150
+
<?php
151
+
152
+
namespace App\AtpRecords;
153
+
154
+
use Carbon\Carbon;
155
+
use SocialDept\AtpParity\Data\Record;
156
+
157
+
class Article extends Record
158
+
{
159
+
public function __construct(
160
+
public readonly string $title,
161
+
public readonly string $body,
162
+
public readonly Carbon $publishedAt,
163
+
public readonly ?array $tags = null,
164
+
public readonly ?string $coverImage = null,
165
+
) {}
166
+
167
+
public static function getLexicon(): string
168
+
{
169
+
return 'com.myapp.blog.article'; // Your custom NSID
170
+
}
171
+
172
+
public static function fromArray(array $data): static
173
+
{
174
+
return new static(
175
+
title: $data['title'],
176
+
body: $data['body'],
177
+
publishedAt: Carbon::parse($data['publishedAt']),
178
+
tags: $data['tags'] ?? null,
179
+
coverImage: $data['coverImage'] ?? null,
180
+
);
181
+
}
182
+
183
+
public function toArray(): array
184
+
{
185
+
return array_filter([
186
+
'$type' => static::getLexicon(),
187
+
'title' => $this->title,
188
+
'body' => $this->body,
189
+
'publishedAt' => $this->publishedAt->toIso8601String(),
190
+
'tags' => $this->tags,
191
+
'coverImage' => $this->coverImage,
192
+
], fn($v) => $v !== null);
193
+
}
194
+
}
195
+
```
196
+
197
+
## Working with Embedded Types
198
+
199
+
atp-schema generates classes for embedded types. Use them in your mappings:
200
+
201
+
```php
202
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
203
+
use SocialDept\AtpSchema\Generated\App\Bsky\Embed\Images;
204
+
use SocialDept\AtpSchema\Generated\App\Bsky\Embed\External;
205
+
use SocialDept\AtpSchema\Generated\Com\Atproto\Repo\StrongRef;
206
+
207
+
$mapper = new SchemaMapper(
208
+
schemaClass: Post::class,
209
+
modelClass: \App\Models\Post::class,
210
+
toAttributes: fn(Post $post) => [
211
+
'content' => $post->text,
212
+
'published_at' => $post->createdAt,
213
+
'has_images' => $post->embed instanceof Images,
214
+
'has_link' => $post->embed instanceof External,
215
+
'embed_data' => $post->embed?->toArray(),
216
+
],
217
+
toRecordData: fn($model) => [
218
+
'text' => $model->content,
219
+
'createdAt' => $model->published_at->toIso8601String(),
220
+
],
221
+
);
222
+
```
223
+
224
+
## Handling Union Types
225
+
226
+
AT Protocol uses union types for fields like `embed`. atp-schema handles these via discriminated unions:
227
+
228
+
```php
229
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
230
+
use SocialDept\AtpSchema\Generated\App\Bsky\Embed\Images;
231
+
use SocialDept\AtpSchema\Generated\App\Bsky\Embed\External;
232
+
use SocialDept\AtpSchema\Generated\App\Bsky\Embed\Record;
233
+
use SocialDept\AtpSchema\Generated\App\Bsky\Embed\RecordWithMedia;
234
+
235
+
$toAttributes = function(Post $post): array {
236
+
$attributes = [
237
+
'content' => $post->text,
238
+
'published_at' => $post->createdAt,
239
+
];
240
+
241
+
// Handle embed union type
242
+
if ($post->embed) {
243
+
match (true) {
244
+
$post->embed instanceof Images => $attributes['embed_type'] = 'images',
245
+
$post->embed instanceof External => $attributes['embed_type'] = 'external',
246
+
$post->embed instanceof Record => $attributes['embed_type'] = 'quote',
247
+
$post->embed instanceof RecordWithMedia => $attributes['embed_type'] = 'quote_media',
248
+
default => $attributes['embed_type'] = 'unknown',
249
+
};
250
+
$attributes['embed_data'] = $post->embed->toArray();
251
+
}
252
+
253
+
return $attributes;
254
+
};
255
+
```
256
+
257
+
## Reply Threading
258
+
259
+
Posts can be replies to other posts:
260
+
261
+
```php
262
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
263
+
264
+
$toAttributes = function(Post $post): array {
265
+
$attributes = [
266
+
'content' => $post->text,
267
+
'published_at' => $post->createdAt,
268
+
'is_reply' => $post->reply !== null,
269
+
];
270
+
271
+
if ($post->reply) {
272
+
// Parent is the immediate post being replied to
273
+
$attributes['reply_parent_uri'] = $post->reply->parent->uri;
274
+
$attributes['reply_parent_cid'] = $post->reply->parent->cid;
275
+
276
+
// Root is the top of the thread
277
+
$attributes['reply_root_uri'] = $post->reply->root->uri;
278
+
$attributes['reply_root_cid'] = $post->reply->root->cid;
279
+
}
280
+
281
+
return $attributes;
282
+
};
283
+
```
284
+
285
+
## Facets (Rich Text)
286
+
287
+
Posts with mentions, links, and hashtags use facets:
288
+
289
+
```php
290
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
291
+
use SocialDept\AtpSchema\Generated\App\Bsky\Richtext\Facet;
292
+
293
+
$toAttributes = function(Post $post): array {
294
+
$attributes = [
295
+
'content' => $post->text,
296
+
'published_at' => $post->createdAt,
297
+
];
298
+
299
+
// Extract mentions, links, and tags from facets
300
+
$mentions = [];
301
+
$links = [];
302
+
$tags = [];
303
+
304
+
foreach ($post->facets ?? [] as $facet) {
305
+
foreach ($facet->features as $feature) {
306
+
$type = $feature->getType();
307
+
match ($type) {
308
+
'app.bsky.richtext.facet#mention' => $mentions[] = $feature->did,
309
+
'app.bsky.richtext.facet#link' => $links[] = $feature->uri,
310
+
'app.bsky.richtext.facet#tag' => $tags[] = $feature->tag,
311
+
default => null,
312
+
};
313
+
}
314
+
}
315
+
316
+
$attributes['mentions'] = $mentions;
317
+
$attributes['links'] = $links;
318
+
$attributes['tags'] = $tags;
319
+
$attributes['facets'] = $post->facets; // Store raw for reconstruction
320
+
321
+
return $attributes;
322
+
};
323
+
```
324
+
325
+
## Type Safety Benefits
326
+
327
+
Using atp-schema classes provides:
328
+
329
+
1. **IDE Autocompletion** - Full property and method suggestions
330
+
2. **Type Checking** - Static analysis catches errors
331
+
3. **Validation** - Data is validated on construction
332
+
4. **Documentation** - Generated classes include docblocks
333
+
334
+
```php
335
+
// IDE knows $post->text is string, $post->createdAt is string, etc.
336
+
$toAttributes = function(Post $post): array {
337
+
return [
338
+
'content' => $post->text, // string
339
+
'published_at' => $post->createdAt, // string (ISO 8601)
340
+
'langs' => $post->langs, // ?array
341
+
'facets' => $post->facets, // ?array
342
+
];
343
+
};
344
+
```
345
+
346
+
## Regenerating Schema Classes
347
+
348
+
When the AT Protocol schema updates, regenerate the classes:
349
+
350
+
```bash
351
+
# In the atp-schema package
352
+
php artisan atp:generate
353
+
```
354
+
355
+
Your mappers will automatically work with the updated types.
+491
docs/atp-signals-integration.md
+491
docs/atp-signals-integration.md
···
1
+
# atp-signals Integration
2
+
3
+
Parity integrates with atp-signals to automatically sync firehose events to your Eloquent models in real-time. The `ParitySignal` class handles create, update, and delete operations for all registered mappers.
4
+
5
+
## ParitySignal
6
+
7
+
The `ParitySignal` is a pre-built signal that listens for commit events and syncs them to your database using your registered mappers.
8
+
9
+
### How It Works
10
+
11
+
1. ParitySignal listens for `commit` events on the firehose
12
+
2. It filters for collections that have registered mappers
13
+
3. For each matching event:
14
+
- **Create/Update**: Upserts the record to your database
15
+
- **Delete**: Removes the record from your database
16
+
17
+
### Setup
18
+
19
+
Register the signal in your atp-signals config:
20
+
21
+
```php
22
+
// config/signal.php
23
+
return [
24
+
'signals' => [
25
+
\SocialDept\AtpParity\Signals\ParitySignal::class,
26
+
],
27
+
];
28
+
```
29
+
30
+
Then start consuming:
31
+
32
+
```bash
33
+
php artisan signal:consume
34
+
```
35
+
36
+
That's it. Your models will automatically sync with the firehose.
37
+
38
+
## What Gets Synced
39
+
40
+
ParitySignal only syncs collections that have registered mappers:
41
+
42
+
```php
43
+
// config/parity.php
44
+
return [
45
+
'mappers' => [
46
+
App\AtpMappers\PostMapper::class, // app.bsky.feed.post
47
+
App\AtpMappers\LikeMapper::class, // app.bsky.feed.like
48
+
App\AtpMappers\FollowMapper::class, // app.bsky.graph.follow
49
+
],
50
+
];
51
+
```
52
+
53
+
With this config, ParitySignal will sync posts, likes, and follows. All other collections are ignored.
54
+
55
+
## Event Flow
56
+
57
+
```
58
+
Firehose Event
59
+
↓
60
+
ParitySignal.handle()
61
+
↓
62
+
Check: Is collection registered?
63
+
↓
64
+
Yes → Get mapper for collection
65
+
↓
66
+
Create DTO from event record
67
+
↓
68
+
Call mapper.upsert() or mapper.deleteByUri()
69
+
↓
70
+
Model saved to database
71
+
```
72
+
73
+
## Example: Syncing Posts
74
+
75
+
### 1. Create the Model
76
+
77
+
```php
78
+
// app/Models/Post.php
79
+
namespace App\Models;
80
+
81
+
use Illuminate\Database\Eloquent\Model;
82
+
use SocialDept\AtpParity\Concerns\SyncsWithAtp;
83
+
84
+
class Post extends Model
85
+
{
86
+
use SyncsWithAtp;
87
+
88
+
protected $fillable = [
89
+
'content',
90
+
'author_did',
91
+
'published_at',
92
+
'atp_uri',
93
+
'atp_cid',
94
+
'atp_synced_at',
95
+
];
96
+
97
+
protected $casts = [
98
+
'published_at' => 'datetime',
99
+
'atp_synced_at' => 'datetime',
100
+
];
101
+
}
102
+
```
103
+
104
+
### 2. Create the Migration
105
+
106
+
```php
107
+
Schema::create('posts', function (Blueprint $table) {
108
+
$table->id();
109
+
$table->text('content');
110
+
$table->string('author_did');
111
+
$table->timestamp('published_at');
112
+
$table->string('atp_uri')->unique();
113
+
$table->string('atp_cid');
114
+
$table->timestamp('atp_synced_at')->nullable();
115
+
$table->timestamps();
116
+
});
117
+
```
118
+
119
+
### 3. Create the Mapper
120
+
121
+
```php
122
+
// app/AtpMappers/PostMapper.php
123
+
namespace App\AtpMappers;
124
+
125
+
use App\Models\Post;
126
+
use Illuminate\Database\Eloquent\Model;
127
+
use SocialDept\AtpParity\RecordMapper;
128
+
use SocialDept\AtpSchema\Data\Data;
129
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post as PostRecord;
130
+
131
+
class PostMapper extends RecordMapper
132
+
{
133
+
public function recordClass(): string
134
+
{
135
+
return PostRecord::class;
136
+
}
137
+
138
+
public function modelClass(): string
139
+
{
140
+
return Post::class;
141
+
}
142
+
143
+
protected function recordToAttributes(Data $record): array
144
+
{
145
+
return [
146
+
'content' => $record->text,
147
+
'published_at' => $record->createdAt,
148
+
];
149
+
}
150
+
151
+
protected function modelToRecordData(Model $model): array
152
+
{
153
+
return [
154
+
'text' => $model->content,
155
+
'createdAt' => $model->published_at->toIso8601String(),
156
+
];
157
+
}
158
+
}
159
+
```
160
+
161
+
### 4. Register Everything
162
+
163
+
```php
164
+
// config/parity.php
165
+
return [
166
+
'mappers' => [
167
+
App\AtpMappers\PostMapper::class,
168
+
],
169
+
];
170
+
```
171
+
172
+
```php
173
+
// config/signal.php
174
+
return [
175
+
'signals' => [
176
+
\SocialDept\AtpParity\Signals\ParitySignal::class,
177
+
],
178
+
];
179
+
```
180
+
181
+
### 5. Start Syncing
182
+
183
+
```bash
184
+
php artisan signal:consume
185
+
```
186
+
187
+
Every new post on the AT Protocol network will now be saved to your `posts` table.
188
+
189
+
## Filtering by User
190
+
191
+
To only sync records from specific users, create a custom signal:
192
+
193
+
```php
194
+
namespace App\Signals;
195
+
196
+
use SocialDept\AtpParity\Signals\ParitySignal;
197
+
use SocialDept\AtpSignals\Events\SignalEvent;
198
+
199
+
class FilteredParitySignal extends ParitySignal
200
+
{
201
+
/**
202
+
* DIDs to sync.
203
+
*/
204
+
protected array $allowedDids = [
205
+
'did:plc:abc123',
206
+
'did:plc:def456',
207
+
];
208
+
209
+
public function handle(SignalEvent $event): void
210
+
{
211
+
// Only process events from allowed DIDs
212
+
if (!in_array($event->did, $this->allowedDids)) {
213
+
return;
214
+
}
215
+
216
+
parent::handle($event);
217
+
}
218
+
}
219
+
```
220
+
221
+
Register your custom signal instead:
222
+
223
+
```php
224
+
// config/signal.php
225
+
return [
226
+
'signals' => [
227
+
App\Signals\FilteredParitySignal::class,
228
+
],
229
+
];
230
+
```
231
+
232
+
## Filtering by Collection
233
+
234
+
To only sync specific collections (even if more mappers are registered):
235
+
236
+
```php
237
+
namespace App\Signals;
238
+
239
+
use SocialDept\AtpParity\Signals\ParitySignal;
240
+
241
+
class PostsOnlySignal extends ParitySignal
242
+
{
243
+
public function collections(): ?array
244
+
{
245
+
// Only sync posts, ignore other registered mappers
246
+
return ['app.bsky.feed.post'];
247
+
}
248
+
}
249
+
```
250
+
251
+
## Custom Processing
252
+
253
+
Add custom logic before or after syncing:
254
+
255
+
```php
256
+
namespace App\Signals;
257
+
258
+
use SocialDept\AtpParity\Contracts\RecordMapper;
259
+
use SocialDept\AtpParity\Signals\ParitySignal;
260
+
use SocialDept\AtpSignals\Events\SignalEvent;
261
+
262
+
class CustomParitySignal extends ParitySignal
263
+
{
264
+
protected function handleUpsert(SignalEvent $event, RecordMapper $mapper): void
265
+
{
266
+
// Pre-processing
267
+
logger()->info('Syncing record', [
268
+
'did' => $event->did,
269
+
'collection' => $event->commit->collection,
270
+
'rkey' => $event->commit->rkey,
271
+
]);
272
+
273
+
// Call parent to do the actual sync
274
+
parent::handleUpsert($event, $mapper);
275
+
276
+
// Post-processing
277
+
// e.g., dispatch a job, send notification, etc.
278
+
}
279
+
280
+
protected function handleDelete(SignalEvent $event, RecordMapper $mapper): void
281
+
{
282
+
logger()->info('Deleting record', [
283
+
'uri' => $this->buildUri($event->did, $event->commit->collection, $event->commit->rkey),
284
+
]);
285
+
286
+
parent::handleDelete($event, $mapper);
287
+
}
288
+
}
289
+
```
290
+
291
+
## Queue Integration
292
+
293
+
For high-volume processing, enable queue mode:
294
+
295
+
```php
296
+
namespace App\Signals;
297
+
298
+
use SocialDept\AtpParity\Signals\ParitySignal;
299
+
300
+
class QueuedParitySignal extends ParitySignal
301
+
{
302
+
public function shouldQueue(): bool
303
+
{
304
+
return true;
305
+
}
306
+
307
+
public function queue(): string
308
+
{
309
+
return 'parity-sync';
310
+
}
311
+
}
312
+
```
313
+
314
+
Then run a dedicated queue worker:
315
+
316
+
```bash
317
+
php artisan queue:work --queue=parity-sync
318
+
```
319
+
320
+
## Multiple Signals
321
+
322
+
You can run ParitySignal alongside other signals:
323
+
324
+
```php
325
+
// config/signal.php
326
+
return [
327
+
'signals' => [
328
+
// Sync to database
329
+
\SocialDept\AtpParity\Signals\ParitySignal::class,
330
+
331
+
// Your custom analytics signal
332
+
App\Signals\AnalyticsSignal::class,
333
+
334
+
// Your moderation signal
335
+
App\Signals\ModerationSignal::class,
336
+
],
337
+
];
338
+
```
339
+
340
+
## Handling High Volume
341
+
342
+
The AT Protocol firehose processes thousands of events per second. For production:
343
+
344
+
### 1. Use Jetstream Mode
345
+
346
+
Jetstream filters server-side, reducing bandwidth:
347
+
348
+
```php
349
+
// config/signal.php
350
+
return [
351
+
'mode' => 'jetstream', // More efficient than firehose
352
+
353
+
'jetstream' => [
354
+
'collections' => [
355
+
'app.bsky.feed.post',
356
+
'app.bsky.feed.like',
357
+
],
358
+
],
359
+
];
360
+
```
361
+
362
+
### 2. Enable Queues
363
+
364
+
Process events asynchronously:
365
+
366
+
```php
367
+
class QueuedParitySignal extends ParitySignal
368
+
{
369
+
public function shouldQueue(): bool
370
+
{
371
+
return true;
372
+
}
373
+
}
374
+
```
375
+
376
+
### 3. Use Database Transactions
377
+
378
+
Batch inserts for better performance:
379
+
380
+
```php
381
+
namespace App\Signals;
382
+
383
+
use Illuminate\Support\Facades\DB;
384
+
use SocialDept\AtpParity\Signals\ParitySignal;
385
+
use SocialDept\AtpSignals\Events\SignalEvent;
386
+
387
+
class BatchedParitySignal extends ParitySignal
388
+
{
389
+
protected array $buffer = [];
390
+
protected int $batchSize = 100;
391
+
392
+
public function handle(SignalEvent $event): void
393
+
{
394
+
$this->buffer[] = $event;
395
+
396
+
if (count($this->buffer) >= $this->batchSize) {
397
+
$this->flush();
398
+
}
399
+
}
400
+
401
+
protected function flush(): void
402
+
{
403
+
DB::transaction(function () {
404
+
foreach ($this->buffer as $event) {
405
+
parent::handle($event);
406
+
}
407
+
});
408
+
409
+
$this->buffer = [];
410
+
}
411
+
}
412
+
```
413
+
414
+
### 4. Monitor Performance
415
+
416
+
Log sync statistics:
417
+
418
+
```php
419
+
namespace App\Signals;
420
+
421
+
use SocialDept\AtpParity\Signals\ParitySignal;
422
+
use SocialDept\AtpSignals\Events\SignalEvent;
423
+
424
+
class MonitoredParitySignal extends ParitySignal
425
+
{
426
+
protected int $processed = 0;
427
+
protected float $startTime;
428
+
429
+
public function handle(SignalEvent $event): void
430
+
{
431
+
$this->startTime ??= microtime(true);
432
+
433
+
parent::handle($event);
434
+
435
+
$this->processed++;
436
+
437
+
if ($this->processed % 1000 === 0) {
438
+
$elapsed = microtime(true) - $this->startTime;
439
+
$rate = $this->processed / $elapsed;
440
+
441
+
logger()->info("Parity sync stats", [
442
+
'processed' => $this->processed,
443
+
'elapsed' => round($elapsed, 2),
444
+
'rate' => round($rate, 2) . '/sec',
445
+
]);
446
+
}
447
+
}
448
+
}
449
+
```
450
+
451
+
## Cursor Management
452
+
453
+
atp-signals handles cursor persistence automatically. If the consumer restarts, it resumes from where it left off.
454
+
455
+
To reset and start fresh:
456
+
457
+
```bash
458
+
php artisan signal:consume --reset
459
+
```
460
+
461
+
## Testing
462
+
463
+
Test your sync setup without connecting to the firehose:
464
+
465
+
```php
466
+
use App\AtpMappers\PostMapper;
467
+
use SocialDept\AtpParity\MapperRegistry;
468
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
469
+
470
+
// Create a test record
471
+
$record = Post::fromArray([
472
+
'text' => 'Test post content',
473
+
'createdAt' => now()->toIso8601String(),
474
+
]);
475
+
476
+
// Get the mapper
477
+
$registry = app(MapperRegistry::class);
478
+
$mapper = $registry->forLexicon('app.bsky.feed.post');
479
+
480
+
// Simulate a sync
481
+
$model = $mapper->upsert($record, [
482
+
'uri' => 'at://did:plc:test/app.bsky.feed.post/test123',
483
+
'cid' => 'bafyretest...',
484
+
]);
485
+
486
+
// Assert
487
+
$this->assertDatabaseHas('posts', [
488
+
'content' => 'Test post content',
489
+
'atp_uri' => 'at://did:plc:test/app.bsky.feed.post/test123',
490
+
]);
491
+
```
+359
docs/importing.md
+359
docs/importing.md
···
1
+
# Importing Records
2
+
3
+
Parity includes a comprehensive import system that enables you to sync historical AT Protocol data to your Eloquent models. This complements the real-time sync provided by [ParitySignal](atp-signals-integration.md).
4
+
5
+
## The Cold Start Problem
6
+
7
+
When you start consuming the AT Protocol firehose with ParitySignal, you only receive events from that point forward. Any records created before you started listening are not captured.
8
+
9
+
Importing solves this "cold start" problem by fetching existing records from user repositories via the `com.atproto.repo.listRecords` API.
10
+
11
+
## Quick Start
12
+
13
+
### 1. Run the Migration
14
+
15
+
Publish and run the migration to create the import state tracking table:
16
+
17
+
```bash
18
+
php artisan vendor:publish --tag=parity-migrations
19
+
php artisan migrate
20
+
```
21
+
22
+
### 2. Import a User
23
+
24
+
```bash
25
+
# Import all registered collections for a user
26
+
php artisan parity:import did:plc:z72i7hdynmk6r22z27h6tvur
27
+
28
+
# Import a specific collection
29
+
php artisan parity:import did:plc:z72i7hdynmk6r22z27h6tvur --collection=app.bsky.feed.post
30
+
31
+
# Show progress
32
+
php artisan parity:import did:plc:z72i7hdynmk6r22z27h6tvur --progress
33
+
```
34
+
35
+
### 3. Check Status
36
+
37
+
```bash
38
+
# Show all import status
39
+
php artisan parity:import-status
40
+
41
+
# Show status for a specific user
42
+
php artisan parity:import-status did:plc:z72i7hdynmk6r22z27h6tvur
43
+
44
+
# Show only incomplete imports
45
+
php artisan parity:import-status --pending
46
+
```
47
+
48
+
## Programmatic Usage
49
+
50
+
### ImportService
51
+
52
+
The `ImportService` is the main orchestration class:
53
+
54
+
```php
55
+
use SocialDept\AtpParity\Import\ImportService;
56
+
57
+
$service = app(ImportService::class);
58
+
59
+
// Import all registered collections for a user
60
+
$result = $service->importUser('did:plc:z72i7hdynmk6r22z27h6tvur');
61
+
62
+
echo "Synced {$result->recordsSynced} records";
63
+
64
+
// Import a specific collection
65
+
$result = $service->importUserCollection(
66
+
'did:plc:z72i7hdynmk6r22z27h6tvur',
67
+
'app.bsky.feed.post'
68
+
);
69
+
70
+
// With progress callback
71
+
$result = $service->importUser('did:plc:z72i7hdynmk6r22z27h6tvur', null, function ($progress) {
72
+
echo "Synced {$progress->recordsSynced} records from {$progress->collection}\n";
73
+
});
74
+
```
75
+
76
+
### ImportResult
77
+
78
+
The `ImportResult` value object provides information about the import operation:
79
+
80
+
```php
81
+
$result = $service->importUser($did);
82
+
83
+
$result->recordsSynced; // Number of records successfully synced
84
+
$result->recordsSkipped; // Number of records skipped
85
+
$result->recordsFailed; // Number of records that failed to sync
86
+
$result->completed; // Whether the import completed fully
87
+
$result->cursor; // Cursor for resuming (if incomplete)
88
+
$result->error; // Error message (if failed)
89
+
90
+
$result->isSuccess(); // True if completed without errors
91
+
$result->isPartial(); // True if some records were synced before failure
92
+
$result->isFailed(); // True if an error occurred
93
+
```
94
+
95
+
### Checking Status
96
+
97
+
```php
98
+
// Check if a collection has been imported
99
+
if ($service->isImported($did, 'app.bsky.feed.post')) {
100
+
echo "Already imported!";
101
+
}
102
+
103
+
// Get detailed status
104
+
$state = $service->getStatus($did, 'app.bsky.feed.post');
105
+
106
+
if ($state) {
107
+
echo "Status: {$state->status}";
108
+
echo "Records synced: {$state->records_synced}";
109
+
}
110
+
111
+
// Get all statuses for a user
112
+
$states = $service->getStatusForUser($did);
113
+
```
114
+
115
+
### Resuming Interrupted Imports
116
+
117
+
If an import is interrupted (network error, timeout, etc.), you can resume it:
118
+
119
+
```php
120
+
// Resume a specific import
121
+
$state = $service->getStatus($did, $collection);
122
+
if ($state && $state->canResume()) {
123
+
$result = $service->resume($state);
124
+
}
125
+
126
+
// Resume all interrupted imports
127
+
$results = $service->resumeAll();
128
+
```
129
+
130
+
### Resetting Import State
131
+
132
+
To re-import a user or collection:
133
+
134
+
```php
135
+
// Reset a specific collection
136
+
$service->reset($did, 'app.bsky.feed.post');
137
+
138
+
// Reset all collections for a user
139
+
$service->resetUser($did);
140
+
```
141
+
142
+
## Queue Integration
143
+
144
+
For large-scale importing, use the queue system:
145
+
146
+
### Command Line
147
+
148
+
```bash
149
+
# Queue an import job instead of running synchronously
150
+
php artisan parity:import did:plc:z72i7hdynmk6r22z27h6tvur --queue
151
+
152
+
# Queue imports for a list of DIDs
153
+
php artisan parity:import --file=dids.txt --queue
154
+
```
155
+
156
+
### Programmatic
157
+
158
+
```php
159
+
use SocialDept\AtpParity\Jobs\ImportUserJob;
160
+
161
+
// Dispatch a single user import
162
+
ImportUserJob::dispatch('did:plc:z72i7hdynmk6r22z27h6tvur');
163
+
164
+
// Dispatch for a specific collection
165
+
ImportUserJob::dispatch('did:plc:z72i7hdynmk6r22z27h6tvur', 'app.bsky.feed.post');
166
+
```
167
+
168
+
## Events
169
+
170
+
Parity dispatches events during importing that you can listen to:
171
+
172
+
### ImportStarted
173
+
174
+
Fired when an import operation begins:
175
+
176
+
```php
177
+
use SocialDept\AtpParity\Events\ImportStarted;
178
+
179
+
Event::listen(ImportStarted::class, function (ImportStarted $event) {
180
+
Log::info("Starting import", [
181
+
'did' => $event->did,
182
+
'collection' => $event->collection,
183
+
]);
184
+
});
185
+
```
186
+
187
+
### ImportProgress
188
+
189
+
Fired after each page of records is processed:
190
+
191
+
```php
192
+
use SocialDept\AtpParity\Events\ImportProgress;
193
+
194
+
Event::listen(ImportProgress::class, function (ImportProgress $event) {
195
+
Log::info("Import progress", [
196
+
'did' => $event->did,
197
+
'collection' => $event->collection,
198
+
'records_synced' => $event->recordsSynced,
199
+
]);
200
+
});
201
+
```
202
+
203
+
### ImportCompleted
204
+
205
+
Fired when an import operation completes successfully:
206
+
207
+
```php
208
+
use SocialDept\AtpParity\Events\ImportCompleted;
209
+
210
+
Event::listen(ImportCompleted::class, function (ImportCompleted $event) {
211
+
$result = $event->result;
212
+
213
+
Log::info("Import completed", [
214
+
'did' => $result->did,
215
+
'collection' => $result->collection,
216
+
'records_synced' => $result->recordsSynced,
217
+
]);
218
+
});
219
+
```
220
+
221
+
### ImportFailed
222
+
223
+
Fired when an import operation fails:
224
+
225
+
```php
226
+
use SocialDept\AtpParity\Events\ImportFailed;
227
+
228
+
Event::listen(ImportFailed::class, function (ImportFailed $event) {
229
+
Log::error("Import failed", [
230
+
'did' => $event->did,
231
+
'collection' => $event->collection,
232
+
'error' => $event->error,
233
+
]);
234
+
});
235
+
```
236
+
237
+
## Configuration
238
+
239
+
Configure importing in `config/parity.php`:
240
+
241
+
```php
242
+
'import' => [
243
+
// Records per page when listing from PDS (max 100)
244
+
'page_size' => 100,
245
+
246
+
// Delay between pages in milliseconds (rate limiting)
247
+
'page_delay' => 100,
248
+
249
+
// Queue name for import jobs
250
+
'queue' => 'parity-import',
251
+
252
+
// Database table for storing import state
253
+
'state_table' => 'parity_import_states',
254
+
],
255
+
```
256
+
257
+
## Batch Importing from File
258
+
259
+
Create a file with DIDs (one per line):
260
+
261
+
```text
262
+
did:plc:z72i7hdynmk6r22z27h6tvur
263
+
did:plc:ewvi7nxzyoun6zhxrhs64oiz
264
+
did:plc:ragtjsm2j2vknwkz3zp4oxrd
265
+
```
266
+
267
+
Then run:
268
+
269
+
```bash
270
+
# Synchronous (one at a time)
271
+
php artisan parity:import --file=dids.txt --progress
272
+
273
+
# Queued (parallel via workers)
274
+
php artisan parity:import --file=dids.txt --queue
275
+
```
276
+
277
+
## Coordinating with ParitySignal
278
+
279
+
For a complete sync solution, combine importing with real-time firehose sync:
280
+
281
+
1. **Start the firehose consumer** - Begin receiving live events
282
+
2. **Import historical data** - Fetch existing records
283
+
3. **Continue firehose sync** - New events are handled automatically
284
+
285
+
This ensures no gaps in your data. Records that arrive via firehose while importing will be properly deduplicated by the mapper's `upsert()` method (which uses the AT Protocol URI as the unique key).
286
+
287
+
```php
288
+
// Example: Import a user then subscribe to their updates
289
+
$service->importUser($did);
290
+
291
+
// The firehose consumer (ParitySignal) handles updates automatically
292
+
// as long as it's running with signal:consume
293
+
```
294
+
295
+
## Best Practices
296
+
297
+
### Rate Limiting
298
+
299
+
The `page_delay` config option helps prevent overwhelming PDS servers. For bulk importing, consider:
300
+
301
+
- Using queued jobs to spread load over time
302
+
- Increasing the delay between pages
303
+
- Running during off-peak hours
304
+
305
+
### Error Handling
306
+
307
+
Imports can fail due to:
308
+
- Network errors
309
+
- PDS rate limiting
310
+
- Invalid records
311
+
312
+
The system automatically tracks progress via cursor, allowing you to resume failed imports:
313
+
314
+
```bash
315
+
# Check for failed imports
316
+
php artisan parity:import-status --failed
317
+
318
+
# Resume all failed/interrupted imports
319
+
php artisan parity:import --resume
320
+
```
321
+
322
+
### Monitoring
323
+
324
+
Use the events to build monitoring:
325
+
326
+
```php
327
+
// Track import metrics
328
+
Event::listen(ImportCompleted::class, function (ImportCompleted $event) {
329
+
Metrics::increment('parity.import.completed');
330
+
Metrics::gauge('parity.import.records', $event->result->recordsSynced);
331
+
});
332
+
333
+
Event::listen(ImportFailed::class, function (ImportFailed $event) {
334
+
Metrics::increment('parity.import.failed');
335
+
Alert::send("Import failed for {$event->did}: {$event->error}");
336
+
});
337
+
```
338
+
339
+
## Database Schema
340
+
341
+
The import state table stores progress:
342
+
343
+
| Column | Type | Description |
344
+
|--------|------|-------------|
345
+
| id | bigint | Primary key |
346
+
| did | string | The DID being imported |
347
+
| collection | string | The collection NSID |
348
+
| status | string | pending, in_progress, completed, failed |
349
+
| cursor | string | Pagination cursor for resuming |
350
+
| records_synced | int | Count of successfully synced records |
351
+
| records_skipped | int | Count of skipped records |
352
+
| records_failed | int | Count of failed records |
353
+
| started_at | timestamp | When import started |
354
+
| completed_at | timestamp | When import completed |
355
+
| error | text | Error message if failed |
356
+
| created_at | timestamp | |
357
+
| updated_at | timestamp | |
358
+
359
+
The combination of `did` and `collection` is unique.
+375
docs/mappers.md
+375
docs/mappers.md
···
1
+
# Record Mappers
2
+
3
+
Mappers are the core of atp-parity. They define bidirectional transformations between AT Protocol record DTOs and Eloquent models.
4
+
5
+
## Creating a Mapper
6
+
7
+
Extend the `RecordMapper` abstract class and implement the required methods:
8
+
9
+
```php
10
+
<?php
11
+
12
+
namespace App\AtpMappers;
13
+
14
+
use App\Models\Post;
15
+
use Illuminate\Database\Eloquent\Model;
16
+
use SocialDept\AtpParity\RecordMapper;
17
+
use SocialDept\AtpSchema\Data\Data;
18
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post as PostRecord;
19
+
20
+
/**
21
+
* @extends RecordMapper<PostRecord, Post>
22
+
*/
23
+
class PostMapper extends RecordMapper
24
+
{
25
+
/**
26
+
* The AT Protocol record class this mapper handles.
27
+
*/
28
+
public function recordClass(): string
29
+
{
30
+
return PostRecord::class;
31
+
}
32
+
33
+
/**
34
+
* The Eloquent model class this mapper handles.
35
+
*/
36
+
public function modelClass(): string
37
+
{
38
+
return Post::class;
39
+
}
40
+
41
+
/**
42
+
* Transform a record DTO into model attributes.
43
+
*/
44
+
protected function recordToAttributes(Data $record): array
45
+
{
46
+
/** @var PostRecord $record */
47
+
return [
48
+
'content' => $record->text,
49
+
'published_at' => $record->createdAt,
50
+
'langs' => $record->langs,
51
+
'facets' => $record->facets,
52
+
];
53
+
}
54
+
55
+
/**
56
+
* Transform a model into record data for creating/updating.
57
+
*/
58
+
protected function modelToRecordData(Model $model): array
59
+
{
60
+
/** @var Post $model */
61
+
return [
62
+
'text' => $model->content,
63
+
'createdAt' => $model->published_at->toIso8601String(),
64
+
'langs' => $model->langs ?? ['en'],
65
+
];
66
+
}
67
+
}
68
+
```
69
+
70
+
## Required Methods
71
+
72
+
### `recordClass(): string`
73
+
74
+
Returns the fully qualified class name of the AT Protocol record DTO. This can be:
75
+
76
+
- A generated class from atp-schema (e.g., `SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post`)
77
+
- A custom class extending `SocialDept\AtpParity\Data\Record`
78
+
79
+
### `modelClass(): string`
80
+
81
+
Returns the fully qualified class name of the Eloquent model.
82
+
83
+
### `recordToAttributes(Data $record): array`
84
+
85
+
Transforms an AT Protocol record into an array of Eloquent model attributes. This is used when:
86
+
87
+
- Creating a new model from a remote record
88
+
- Updating an existing model from a remote record
89
+
90
+
### `modelToRecordData(Model $model): array`
91
+
92
+
Transforms an Eloquent model into an array suitable for creating an AT Protocol record. This is used when:
93
+
94
+
- Publishing a local model to the AT Protocol network
95
+
- Comparing local and remote state
96
+
97
+
## Inherited Methods
98
+
99
+
The abstract `RecordMapper` class provides these methods:
100
+
101
+
### `lexicon(): string`
102
+
103
+
Returns the lexicon NSID (e.g., `app.bsky.feed.post`). Automatically derived from the record class's `getLexicon()` method.
104
+
105
+
### `toModel(Data $record, array $meta = []): Model`
106
+
107
+
Creates a new (unsaved) model instance from a record DTO.
108
+
109
+
```php
110
+
$record = PostRecord::fromArray($data);
111
+
$model = $mapper->toModel($record, [
112
+
'uri' => 'at://did:plc:xxx/app.bsky.feed.post/abc123',
113
+
'cid' => 'bafyre...',
114
+
]);
115
+
```
116
+
117
+
### `toRecord(Model $model): Data`
118
+
119
+
Converts a model back to a record DTO.
120
+
121
+
```php
122
+
$record = $mapper->toRecord($post);
123
+
// Use $record->toArray() to get data for API calls
124
+
```
125
+
126
+
### `updateModel(Model $model, Data $record, array $meta = []): Model`
127
+
128
+
Updates an existing model with data from a record. Does not save the model.
129
+
130
+
```php
131
+
$mapper->updateModel($existingPost, $record, ['cid' => $newCid]);
132
+
$existingPost->save();
133
+
```
134
+
135
+
### `findByUri(string $uri): ?Model`
136
+
137
+
Finds a model by its AT Protocol URI.
138
+
139
+
```php
140
+
$post = $mapper->findByUri('at://did:plc:xxx/app.bsky.feed.post/abc123');
141
+
```
142
+
143
+
### `upsert(Data $record, array $meta = []): Model`
144
+
145
+
Creates or updates a model based on the URI. This is the primary method used for syncing.
146
+
147
+
```php
148
+
$post = $mapper->upsert($record, [
149
+
'uri' => $uri,
150
+
'cid' => $cid,
151
+
]);
152
+
```
153
+
154
+
### `deleteByUri(string $uri): bool`
155
+
156
+
Deletes a model by its AT Protocol URI.
157
+
158
+
```php
159
+
$deleted = $mapper->deleteByUri('at://did:plc:xxx/app.bsky.feed.post/abc123');
160
+
```
161
+
162
+
## Meta Fields
163
+
164
+
The `$meta` array passed to `toModel`, `updateModel`, and `upsert` can contain:
165
+
166
+
| Key | Description |
167
+
|-----|-------------|
168
+
| `uri` | The AT Protocol URI (e.g., `at://did:plc:xxx/app.bsky.feed.post/abc123`) |
169
+
| `cid` | The content identifier hash |
170
+
171
+
These are automatically mapped to your configured column names (default: `atp_uri`, `atp_cid`).
172
+
173
+
## Customizing Column Names
174
+
175
+
Override the column methods to use different database columns:
176
+
177
+
```php
178
+
class PostMapper extends RecordMapper
179
+
{
180
+
protected function uriColumn(): string
181
+
{
182
+
return 'at_uri'; // Instead of default 'atp_uri'
183
+
}
184
+
185
+
protected function cidColumn(): string
186
+
{
187
+
return 'at_cid'; // Instead of default 'atp_cid'
188
+
}
189
+
190
+
// ... other methods
191
+
}
192
+
```
193
+
194
+
Or configure globally in `config/parity.php`:
195
+
196
+
```php
197
+
'columns' => [
198
+
'uri' => 'at_uri',
199
+
'cid' => 'at_cid',
200
+
],
201
+
```
202
+
203
+
## Registering Mappers
204
+
205
+
### Via Configuration
206
+
207
+
Add your mapper classes to `config/parity.php`:
208
+
209
+
```php
210
+
return [
211
+
'mappers' => [
212
+
App\AtpMappers\PostMapper::class,
213
+
App\AtpMappers\ProfileMapper::class,
214
+
App\AtpMappers\LikeMapper::class,
215
+
],
216
+
];
217
+
```
218
+
219
+
### Programmatically
220
+
221
+
Register mappers at runtime via the `MapperRegistry`:
222
+
223
+
```php
224
+
use SocialDept\AtpParity\MapperRegistry;
225
+
226
+
$registry = app(MapperRegistry::class);
227
+
$registry->register(new PostMapper());
228
+
```
229
+
230
+
## Using the Registry
231
+
232
+
The `MapperRegistry` provides lookup methods:
233
+
234
+
```php
235
+
use SocialDept\AtpParity\MapperRegistry;
236
+
237
+
$registry = app(MapperRegistry::class);
238
+
239
+
// Find mapper by record class
240
+
$mapper = $registry->forRecord(PostRecord::class);
241
+
242
+
// Find mapper by model class
243
+
$mapper = $registry->forModel(Post::class);
244
+
245
+
// Find mapper by lexicon NSID
246
+
$mapper = $registry->forLexicon('app.bsky.feed.post');
247
+
248
+
// Get all registered lexicons
249
+
$lexicons = $registry->lexicons();
250
+
// ['app.bsky.feed.post', 'app.bsky.actor.profile', ...]
251
+
```
252
+
253
+
## SchemaMapper for Quick Setup
254
+
255
+
For simple mappings, use `SchemaMapper` instead of creating a full class:
256
+
257
+
```php
258
+
use SocialDept\AtpParity\Support\SchemaMapper;
259
+
use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Like;
260
+
261
+
$mapper = new SchemaMapper(
262
+
schemaClass: Like::class,
263
+
modelClass: \App\Models\Like::class,
264
+
toAttributes: fn(Like $like) => [
265
+
'subject_uri' => $like->subject->uri,
266
+
'subject_cid' => $like->subject->cid,
267
+
'liked_at' => $like->createdAt,
268
+
],
269
+
toRecordData: fn($model) => [
270
+
'subject' => [
271
+
'uri' => $model->subject_uri,
272
+
'cid' => $model->subject_cid,
273
+
],
274
+
'createdAt' => $model->liked_at->toIso8601String(),
275
+
],
276
+
);
277
+
278
+
$registry->register($mapper);
279
+
```
280
+
281
+
## Handling Complex Records
282
+
283
+
### Embedded Objects
284
+
285
+
AT Protocol records often contain embedded objects. Handle them in your mapping:
286
+
287
+
```php
288
+
protected function recordToAttributes(Data $record): array
289
+
{
290
+
/** @var PostRecord $record */
291
+
$attributes = [
292
+
'content' => $record->text,
293
+
'published_at' => $record->createdAt,
294
+
];
295
+
296
+
// Handle reply reference
297
+
if ($record->reply) {
298
+
$attributes['reply_to_uri'] = $record->reply->parent->uri;
299
+
$attributes['thread_root_uri'] = $record->reply->root->uri;
300
+
}
301
+
302
+
// Handle embed
303
+
if ($record->embed) {
304
+
$attributes['embed_type'] = $record->embed->getType();
305
+
$attributes['embed_data'] = $record->embed->toArray();
306
+
}
307
+
308
+
return $attributes;
309
+
}
310
+
```
311
+
312
+
### Facets (Rich Text)
313
+
314
+
Posts with mentions, links, and hashtags have facets:
315
+
316
+
```php
317
+
protected function recordToAttributes(Data $record): array
318
+
{
319
+
/** @var PostRecord $record */
320
+
return [
321
+
'content' => $record->text,
322
+
'facets' => $record->facets, // Store as JSON
323
+
'published_at' => $record->createdAt,
324
+
];
325
+
}
326
+
327
+
protected function modelToRecordData(Model $model): array
328
+
{
329
+
/** @var Post $model */
330
+
return [
331
+
'text' => $model->content,
332
+
'facets' => $model->facets, // Restore from JSON
333
+
'createdAt' => $model->published_at->toIso8601String(),
334
+
];
335
+
}
336
+
```
337
+
338
+
## Multiple Mappers per Lexicon
339
+
340
+
You can register multiple mappers for different model types:
341
+
342
+
```php
343
+
// Map posts to different models based on criteria
344
+
class UserPostMapper extends RecordMapper
345
+
{
346
+
public function recordClass(): string
347
+
{
348
+
return PostRecord::class;
349
+
}
350
+
351
+
public function modelClass(): string
352
+
{
353
+
return UserPost::class;
354
+
}
355
+
356
+
// ... mapping logic for user's own posts
357
+
}
358
+
359
+
class FeedPostMapper extends RecordMapper
360
+
{
361
+
public function recordClass(): string
362
+
{
363
+
return PostRecord::class;
364
+
}
365
+
366
+
public function modelClass(): string
367
+
{
368
+
return FeedPost::class;
369
+
}
370
+
371
+
// ... mapping logic for feed posts
372
+
}
373
+
```
374
+
375
+
Note: The registry will return the first registered mapper for a given lexicon. Use explicit mapper instances when you need specific behavior.
+363
docs/traits.md
+363
docs/traits.md
···
1
+
# Model Traits
2
+
3
+
Parity provides two traits to add AT Protocol awareness to your Eloquent models.
4
+
5
+
## HasAtpRecord
6
+
7
+
The base trait for models that store AT Protocol record references.
8
+
9
+
### Setup
10
+
11
+
```php
12
+
<?php
13
+
14
+
namespace App\Models;
15
+
16
+
use Illuminate\Database\Eloquent\Model;
17
+
use SocialDept\AtpParity\Concerns\HasAtpRecord;
18
+
19
+
class Post extends Model
20
+
{
21
+
use HasAtpRecord;
22
+
23
+
protected $fillable = [
24
+
'content',
25
+
'published_at',
26
+
'atp_uri',
27
+
'atp_cid',
28
+
];
29
+
}
30
+
```
31
+
32
+
### Database Migration
33
+
34
+
```php
35
+
Schema::create('posts', function (Blueprint $table) {
36
+
$table->id();
37
+
$table->text('content');
38
+
$table->timestamp('published_at');
39
+
$table->string('atp_uri')->nullable()->unique();
40
+
$table->string('atp_cid')->nullable();
41
+
$table->timestamps();
42
+
});
43
+
```
44
+
45
+
### Available Methods
46
+
47
+
#### `getAtpUri(): ?string`
48
+
49
+
Returns the stored AT Protocol URI.
50
+
51
+
```php
52
+
$post->getAtpUri();
53
+
// "at://did:plc:abc123/app.bsky.feed.post/xyz789"
54
+
```
55
+
56
+
#### `getAtpCid(): ?string`
57
+
58
+
Returns the stored content identifier.
59
+
60
+
```php
61
+
$post->getAtpCid();
62
+
// "bafyreib2rxk3rjnlvzj..."
63
+
```
64
+
65
+
#### `getAtpDid(): ?string`
66
+
67
+
Extracts the DID from the URI.
68
+
69
+
```php
70
+
$post->getAtpDid();
71
+
// "did:plc:abc123"
72
+
```
73
+
74
+
#### `getAtpCollection(): ?string`
75
+
76
+
Extracts the collection (lexicon NSID) from the URI.
77
+
78
+
```php
79
+
$post->getAtpCollection();
80
+
// "app.bsky.feed.post"
81
+
```
82
+
83
+
#### `getAtpRkey(): ?string`
84
+
85
+
Extracts the record key from the URI.
86
+
87
+
```php
88
+
$post->getAtpRkey();
89
+
// "xyz789"
90
+
```
91
+
92
+
#### `hasAtpRecord(): bool`
93
+
94
+
Checks if the model has been synced to AT Protocol.
95
+
96
+
```php
97
+
if ($post->hasAtpRecord()) {
98
+
// Model exists on AT Protocol
99
+
}
100
+
```
101
+
102
+
#### `getAtpMapper(): ?RecordMapper`
103
+
104
+
Gets the registered mapper for this model class.
105
+
106
+
```php
107
+
$mapper = $post->getAtpMapper();
108
+
```
109
+
110
+
#### `toAtpRecord(): ?Data`
111
+
112
+
Converts the model to an AT Protocol record DTO.
113
+
114
+
```php
115
+
$record = $post->toAtpRecord();
116
+
$data = $record->toArray(); // Ready for API calls
117
+
```
118
+
119
+
### Query Scopes
120
+
121
+
#### `scopeWithAtpRecord($query)`
122
+
123
+
Query only models that have been synced.
124
+
125
+
```php
126
+
$syncedPosts = Post::withAtpRecord()->get();
127
+
```
128
+
129
+
#### `scopeWithoutAtpRecord($query)`
130
+
131
+
Query only models that have NOT been synced.
132
+
133
+
```php
134
+
$localOnlyPosts = Post::withoutAtpRecord()->get();
135
+
```
136
+
137
+
#### `scopeWhereAtpUri($query, string $uri)`
138
+
139
+
Find a model by its AT Protocol URI.
140
+
141
+
```php
142
+
$post = Post::whereAtpUri('at://did:plc:xxx/app.bsky.feed.post/abc')->first();
143
+
```
144
+
145
+
## SyncsWithAtp
146
+
147
+
Extended trait for bidirectional synchronization tracking. Includes all `HasAtpRecord` functionality plus sync timestamps and conflict detection.
148
+
149
+
### Setup
150
+
151
+
```php
152
+
<?php
153
+
154
+
namespace App\Models;
155
+
156
+
use Illuminate\Database\Eloquent\Model;
157
+
use SocialDept\AtpParity\Concerns\SyncsWithAtp;
158
+
159
+
class Post extends Model
160
+
{
161
+
use SyncsWithAtp;
162
+
163
+
protected $fillable = [
164
+
'content',
165
+
'published_at',
166
+
'atp_uri',
167
+
'atp_cid',
168
+
'atp_synced_at',
169
+
];
170
+
171
+
protected $casts = [
172
+
'published_at' => 'datetime',
173
+
'atp_synced_at' => 'datetime',
174
+
];
175
+
}
176
+
```
177
+
178
+
### Database Migration
179
+
180
+
```php
181
+
Schema::create('posts', function (Blueprint $table) {
182
+
$table->id();
183
+
$table->text('content');
184
+
$table->timestamp('published_at');
185
+
$table->string('atp_uri')->nullable()->unique();
186
+
$table->string('atp_cid')->nullable();
187
+
$table->timestamp('atp_synced_at')->nullable();
188
+
$table->timestamps();
189
+
});
190
+
```
191
+
192
+
### Additional Methods
193
+
194
+
#### `getAtpSyncedAtColumn(): string`
195
+
196
+
Returns the column name for the sync timestamp. Override to customize.
197
+
198
+
```php
199
+
public function getAtpSyncedAtColumn(): string
200
+
{
201
+
return 'last_synced_at'; // Default: 'atp_synced_at'
202
+
}
203
+
```
204
+
205
+
#### `getAtpSyncedAt(): ?DateTimeInterface`
206
+
207
+
Returns when the model was last synced.
208
+
209
+
```php
210
+
$syncedAt = $post->getAtpSyncedAt();
211
+
// Carbon instance or null
212
+
```
213
+
214
+
#### `markAsSynced(string $uri, string $cid): void`
215
+
216
+
Marks the model as synced with the given metadata. Does not save.
217
+
218
+
```php
219
+
$post->markAsSynced($uri, $cid);
220
+
$post->save();
221
+
```
222
+
223
+
#### `hasLocalChanges(): bool`
224
+
225
+
Checks if the model has been modified since the last sync.
226
+
227
+
```php
228
+
if ($post->hasLocalChanges()) {
229
+
// Local changes exist that haven't been pushed
230
+
}
231
+
```
232
+
233
+
This compares `updated_at` with `atp_synced_at`.
234
+
235
+
#### `updateFromRecord(Data $record, string $uri, string $cid): void`
236
+
237
+
Updates the model from a remote record. Does not save.
238
+
239
+
```php
240
+
$post->updateFromRecord($record, $uri, $cid);
241
+
$post->save();
242
+
```
243
+
244
+
## Practical Examples
245
+
246
+
### Checking Sync Status
247
+
248
+
```php
249
+
$post = Post::find(1);
250
+
251
+
if (!$post->hasAtpRecord()) {
252
+
echo "Not yet published to AT Protocol";
253
+
} elseif ($post->hasLocalChanges()) {
254
+
echo "Has unpushed local changes";
255
+
} else {
256
+
echo "In sync with AT Protocol";
257
+
}
258
+
```
259
+
260
+
### Finding Related Records
261
+
262
+
```php
263
+
// Get all posts from the same author
264
+
$authorDid = $post->getAtpDid();
265
+
$authorPosts = Post::withAtpRecord()
266
+
->get()
267
+
->filter(fn($p) => $p->getAtpDid() === $authorDid);
268
+
```
269
+
270
+
### Building an AT Protocol URL
271
+
272
+
```php
273
+
$post = Post::find(1);
274
+
275
+
if ($post->hasAtpRecord()) {
276
+
$bskyUrl = sprintf(
277
+
'https://bsky.app/profile/%s/post/%s',
278
+
$post->getAtpDid(),
279
+
$post->getAtpRkey()
280
+
);
281
+
}
282
+
```
283
+
284
+
### Sync Status Dashboard
285
+
286
+
```php
287
+
// Get sync statistics
288
+
$stats = [
289
+
'total' => Post::count(),
290
+
'synced' => Post::withAtpRecord()->count(),
291
+
'pending' => Post::withoutAtpRecord()->count(),
292
+
'with_changes' => Post::withAtpRecord()
293
+
->get()
294
+
->filter(fn($p) => $p->hasLocalChanges())
295
+
->count(),
296
+
];
297
+
```
298
+
299
+
## Custom Column Names
300
+
301
+
Both traits respect the global column configuration:
302
+
303
+
```php
304
+
// config/parity.php
305
+
return [
306
+
'columns' => [
307
+
'uri' => 'at_protocol_uri',
308
+
'cid' => 'at_protocol_cid',
309
+
],
310
+
];
311
+
```
312
+
313
+
For the sync timestamp column, override the method in your model:
314
+
315
+
```php
316
+
class Post extends Model
317
+
{
318
+
use SyncsWithAtp;
319
+
320
+
public function getAtpSyncedAtColumn(): string
321
+
{
322
+
return 'last_synced_at';
323
+
}
324
+
}
325
+
```
326
+
327
+
## Event Hooks
328
+
329
+
The `SyncsWithAtp` trait includes a boot method you can extend:
330
+
331
+
```php
332
+
class Post extends Model
333
+
{
334
+
use SyncsWithAtp;
335
+
336
+
protected static function bootSyncsWithAtp(): void
337
+
{
338
+
parent::bootSyncsWithAtp();
339
+
340
+
static::updating(function ($model) {
341
+
// Custom logic before updates
342
+
});
343
+
}
344
+
}
345
+
```
346
+
347
+
## Combining with Other Traits
348
+
349
+
The traits work alongside other Eloquent features:
350
+
351
+
```php
352
+
use Illuminate\Database\Eloquent\Model;
353
+
use Illuminate\Database\Eloquent\SoftDeletes;
354
+
use SocialDept\AtpParity\Concerns\SyncsWithAtp;
355
+
356
+
class Post extends Model
357
+
{
358
+
use SoftDeletes;
359
+
use SyncsWithAtp;
360
+
361
+
// Both traits work together
362
+
}
363
+
```
header.png
header.png
This is a binary file and will not be displayed.