README

Bidirectional mapping between AT Protocol records and Laravel Eloquent models.

What is Parity?

Parity is a Laravel package that bridges your Eloquent models with AT Protocol records. It provides bidirectional mapping, automatic firehose synchronization, and type-safe transformations between your database and the decentralized social web.

Think of it as Laravel's model casts, but for AT Protocol records.

Why use Parity?

Laravel-style code - Familiar patterns you already know
Bidirectional mapping - Transform records to models and back
Firehose sync - Automatically sync network events to your database
Type-safe DTOs - Full integration with atp-schema generated types
Model traits - Add AT Protocol awareness to any Eloquent model
Flexible mappers - Define custom transformations for your domain
Blob handling - Download, upload, and serve images and videos

Quick Example

use SocialDept\AtpParity\Acceptance\Acceptance;
use SocialDept\AtpParity\Fields\Field;
use SocialDept\AtpParity\RecordMapper;

class PostMapper extends RecordMapper
{
    public function recordClass(): string
    {
        return \SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post::class;
    }

    public function modelClass(): string
    {
        return \App\Models\Post::class;
    }

    public function fields(): array
    {
        return [
            'text' => 'content',
            'createdAt' => Field::for('published_at'),
        ];
    }

    public function accepts(): ?Acceptance
    {
        return Acceptance::connectedActors();
    }
}

One declaration gives both directions. It also lets the package answer which columns end up in the record, so a save touching none of them does not write to the author's repo.

accepts() is not optional. A mapper is an ingest boundary, and one that has not said what it allows imports nothing. See Upgrading if you are coming from 0.6 or earlier.

Installation

composer require socialdept/atp-parity

Optionally publish the configuration:

php artisan vendor:publish --tag=atp-parity-config

Getting Started

Once installed, you're three steps away from syncing AT Protocol records:

1. Create a Mapper

Define how your record maps to your model:

class PostMapper extends RecordMapper
{
    public function recordClass(): string
    {
        return Post::class; // Your atp-schema DTO or custom Record
    }

    public function modelClass(): string
    {
        return \App\Models\Post::class;
    }

    public function fields(): array
    {
        return ['text' => 'content'];
    }

    public function accepts(): ?Acceptance
    {
        return Acceptance::connectedActors();
    }
}

A field can name a column directly, carry a default, cast an enum, delegate to a codec, or supply either direction as a closure:

public function fields(): array
{
    return [
        'text' => 'content',
        'preferences.timezone' => Field::for('timezone')->default('UTC'),
        'theme' => Field::for('palette')->codec(ThemeCodec::class)->lossy(),
        'url' => Field::derived(fn ($model) => $model->url()),
        'avatar' => Field::for('avatar')->blob(),
    ];
}

A field is built with Field::for($column), or Field::derived($closure) when it has no column and is written only. Everything else chains: get(), set(), default(), codec(), enum(), lossy(), blob(), importOnly(), overflowsToBlob().

Overflowing a large field to a blob

A record has a hard ceiling of 1 MiB, and the guidance is to keep records within a few dozen KBytes and use a blob beyond that. A field holding something open ended, such as a document body, can declare where to put it once the record no longer fits:

'content.items' => Field::derived(fn ($model) => $model->blocks())
    ->overflowsToBlob(
        blob: 'content.blob',
        references: 'content.references',
    ),

Under the threshold the field is written inline as usual. Over it, the value is uploaded as a blob, the inline path is dropped, and the record is read back the same way on import. This is opt in: a field that does not declare it is never touched.

Pass references wherever the value can contain blobs. A PDS collects a blob that no record references, and a blob nested inside overflowed content is invisible to it, so the images in a long record are swept unless the record keeps naming them.

PARITY_RECORD_OVERFLOW_BYTES sets the default threshold, and a field may override it.

The lexicon has to accept both shapes. Adding the two properties to an existing content object is backwards compatible, since a reader that knows only items keeps working on every record written so far:

{
  "lexicon": 1,
  "id": "com.example.content",
  "defs": {
    "main": {
      "type": "object",
      "properties": {
        "items": {
          "type": "array",
          "items": { "type": "union", "refs": [], "closed": false },
          "description": "Content blocks, written inline while the record fits"
        },
        "blob": {
          "type": "blob",
          "accept": ["application/json"],
          "maxSize": 5242880,
          "description": "The same content as JSON, written here instead of inline once the record is too large"
        },
        "references": {
          "type": "array",
          "items": { "type": "blob" },
          "description": "Every blob the content refers to, re-listed so a PDS does not collect them while the content sits in a blob"
        }
      }
    }
  }
}

Three things to get right:

The inline property cannot stay required. An overflowed record does not carry it, so a lexicon demanding items rejects exactly the records this feature produces.
accept must match the field's mime type. The default is text/plain, so a lexicon accepting only application/json needs overflowsToBlob(..., mimeType: 'application/json') to agree with it.
maxSize bounds the overflow, not the record. It is the ceiling on how much content a blob may hold, well above the record threshold rather than near it.

Where a translation does not fit a declaration, override recordToAttributes() or modelToRecordData() instead. That is still supported and is the right choice for something like rich text.

2. Register Your Mapper

// config/atp-parity.php
return [
    'mappers' => [
        App\AtpMappers\PostMapper::class,
    ],
];

3. Add the Trait to Your Model

use SocialDept\AtpParity\Concerns\HasAtpRecord;

class Post extends Model
{
    use HasAtpRecord;
}

Your model can now convert to/from AT Protocol records and query by URI.

What can you build?

Data mirrors - Keep local copies of AT Protocol data
AppViews - Build custom applications with synced data
Analytics platforms - Store and analyze network activity
Content aggregators - Collect and organize posts locally
Moderation tools - Track and manage content in your database
Hybrid applications - Combine local and federated data

Ecosystem Integration

Parity is designed to work seamlessly with the other atp-* packages:

Package	Integration
atp-schema	Records extend `Data`, use generated DTOs directly
atp-client	`RecordHelper` for fetching and hydrating records
atp-signals	`ParitySignal` for automatic firehose sync

Using with atp-schema

Use generated schema classes directly with SchemaMapper:

use SocialDept\AtpSchema\Generated\App\Bsky\Feed\Post;
use SocialDept\AtpParity\Support\SchemaMapper;

$mapper = new SchemaMapper(
    schemaClass: Post::class,
    modelClass: \App\Models\Post::class,
    toAttributes: fn(Post $p) => [
        'content' => $p->text,
        'published_at' => $p->createdAt,
    ],
    toRecordData: fn($m) => [
        'text' => $m->content,
        'createdAt' => $m->published_at->toIso8601String(),
    ],
);

$registry->register($mapper);

Using with atp-client

Fetch records by URI and convert directly to models:

use SocialDept\AtpParity\Support\RecordHelper;

$helper = app(RecordHelper::class);

// Fetch as typed DTO
$record = $helper->fetch('at://did:plc:xxx/app.bsky.feed.post/abc123');

// Fetch and convert to model (unsaved)
$post = $helper->fetchAsModel('at://did:plc:xxx/app.bsky.feed.post/abc123');

// Fetch and sync to database (upsert)
$post = $helper->sync('at://did:plc:xxx/app.bsky.feed.post/abc123');

The helper automatically resolves the DID to find the correct PDS endpoint, so it works with any AT Protocol server - not just Bluesky.

Using with atp-signals

Enable automatic firehose synchronization by registering the ParitySignal:

// config/signal.php
return [
    'signals' => [
        \SocialDept\AtpParity\Signals\ParitySignal::class,
    ],
];

Run php artisan signal:consume and your models will automatically sync with matching firehose events.

Importing Historical Data

For existing records created before you started consuming the firehose:

# Import a user's records
php artisan parity:import did:plc:z72i7hdynmk6r22z27h6tvur

# Check import status
php artisan parity:import-status

Or programmatically:

use SocialDept\AtpParity\Import\ImportService;

$service = app(ImportService::class);
$result = $service->importUser('did:plc:z72i7hdynmk6r22z27h6tvur');

echo "Synced {$result->recordsSynced} records";

Documentation

For detailed documentation on specific topics:

Record Mappers - Creating and using mappers
Model Traits - HasAtpRecord and SyncsWithAtp
Automatic Syncing - Auto-sync models with AT Protocol
Blob Handling - Downloading, uploading, and serving blobs
atp-schema Integration - Using generated DTOs
atp-client Integration - RecordHelper and fetching
atp-signals Integration - ParitySignal and firehose sync
Importing - Syncing historical data
Upgrading - Breaking changes between versions

Model Traits

HasAtpRecord

Add AT Protocol awareness to your models:

use SocialDept\AtpParity\Concerns\HasAtpRecord;

class Post extends Model
{
    use HasAtpRecord;

    protected $fillable = ['content', 'atp_uri', 'atp_cid'];
}

Available methods:

// Get AT Protocol metadata
$post->getAtpUri();        // at://did:plc:xxx/app.bsky.feed.post/rkey
$post->getAtpCid();        // bafyre...
$post->getAtpDid();        // did:plc:xxx (extracted from URI)
$post->getAtpCollection(); // app.bsky.feed.post (extracted from URI)
$post->getAtpRkey();       // rkey (extracted from URI)

// Check sync status
$post->hasAtpRecord();     // true if synced

// Convert to record DTO
$record = $post->toAtpRecord();

// Query scopes
Post::withAtpRecord()->get();      // Only synced posts
Post::withoutAtpRecord()->get();   // Only unsynced posts
Post::whereAtpUri($uri)->first();  // Find by URI

SyncsWithAtp

Extended trait for bidirectional sync tracking:

use SocialDept\AtpParity\Concerns\SyncsWithAtp;

class Post extends Model
{
    use SyncsWithAtp;
}

Additional methods:

// Track sync status
$post->getAtpSyncedAt();   // Last sync timestamp
$post->hasLocalChanges();  // True if updated since last sync

// Mark as synced
$post->markAsSynced($uri, $cid);

// Update from remote
$post->updateFromRecord($record, $uri, $cid);

HasAtpBlobs

Add blob handling to models with images or other binary content:

use SocialDept\AtpParity\Concerns\HasAtpRecord;
use SocialDept\AtpParity\Concerns\HasAtpBlobs;

class Post extends Model
{
    use HasAtpRecord, HasAtpBlobs;

    protected $casts = ['atp_blobs' => 'array'];
}

Available methods:

// Get URLs for blobs
$url = $post->getAtpBlobUrl('avatar');     // Single blob URL
$urls = $post->getAtpBlobUrls('images');   // Array of URLs

// Download blobs locally
$post->downloadAtpBlobs();

// Check status
$post->hasAtpBlobs();      // Has any blob data
$post->hasLocalBlobs();    // All blobs downloaded locally

See Blob Handling for complete documentation including MediaLibrary integration.

AutoSyncsWithAtp

Automatically sync models with AT Protocol on create, update, and delete:

use SocialDept\AtpParity\Concerns\AutoSyncsWithAtp;

class Post extends Model
{
    use AutoSyncsWithAtp;

    public function syncAsDid(): ?string
    {
        return $this->user->did;
    }

    public function shouldAutoSync(): bool
    {
        return $this->status === 'published';
    }
}

When enabled, the model automatically syncs:

$post = Post::create(['content' => 'Hello!']);   // Syncs to ATP
$post->update(['content' => 'Updated']);         // Updates the record
$post->update(['view_count' => 41]);             // No write: not in the record
$post->delete();                                 // Removes from ATP

An update only syncs when it changed a column the record actually contains, which the mapper's fields() declaration determines. A mapper that overrides its own directions cannot report that, so every update syncs, as before.

Before writing, an unchanged record is skipped by comparing the CID it would have against the one already stored. Set PARITY_SYNC_SKIP_UNCHANGED=false to disable, and pass resync(force: true) when repairing a repo, where an equal CID proves what was last written rather than what the repo still holds.

Failed syncs due to expired OAuth sessions can be captured and retried after re-authentication. See Automatic Syncing for complete documentation including pending sync configuration.

Database Migration

Add AT Protocol columns to your models:

Schema::table('posts', function (Blueprint $table) {
    $table->string('atp_uri')->nullable()->unique();
    $table->string('atp_cid')->nullable();
    $table->timestamp('atp_synced_at')->nullable(); // For SyncsWithAtp
    $table->json('atp_blobs')->nullable();          // For HasAtpBlobs
});

Publish and run Parity's core migration:

php artisan vendor:publish --tag=parity-migrations
php artisan migrate

Optional migrations are published separately based on which features you use:

# Manual conflict resolution (conflicts.strategy = 'manual')
php artisan vendor:publish --tag=parity-migrations-conflicts

# Filesystem blob storage (blobs.storage_driver = 'filesystem')
php artisan vendor:publish --tag=parity-migrations-blobs

# Database pending sync storage (pending_syncs.storage = 'database')
php artisan vendor:publish --tag=parity-migrations-pending-syncs

Configuration

// config/atp-parity.php
return [
    // Registered mappers
    'mappers' => [
        App\AtpMappers\PostMapper::class,
        App\AtpMappers\ProfileMapper::class,
    ],

    // Column names for AT Protocol metadata
    'columns' => [
        'uri' => 'atp_uri',
        'cid' => 'atp_cid',
    ],

    // Two different questions, neither falling back to the other. Without a lookup
    // the policy that needs it accepts nothing.
    'acceptance' => [
        // Do we hold credentials for this repo, so an inbound record may be our own?
        'is_connected_actor' => fn (string $did) => \App\Models\LoginMethod::validFor($did)->exists(),

        // Do we know this actor at all, whether or not we can write for them?
        'is_known_actor' => fn (string $did) => \App\Models\User::where('did', $did)->exists(),
    ],

    // Steps that bring older record shapes up to the current one. Each declares its
    // lexicon and position with #[UpcastsFrom].
    'upcasters' => [
        App\AtpUpcasters\SplitPalette::class,
    ],

    'sync' => [
        // Skip a resync when the repo already holds the record byte for byte.
        'skip_unchanged' => env('PARITY_SYNC_SKIP_UNCHANGED', true),
    ],

    // Blob handling configuration
    'blobs' => [
        // Turns a model's attached file into a blob reference. The only part of
        // writing a record allowed to perform I/O, and it runs before construction.
        'resolver' => App\Atp\MediaBlobResolver::class,

        // 'filesystem' (requires migrations) or 'medialibrary' (no extra migrations)
        'storage_driver' => \SocialDept\AtpParity\Enums\BlobStorageDriver::Filesystem,
        'download_on_import' => env('PARITY_BLOB_DOWNLOAD', false),
        'disk' => env('PARITY_BLOB_DISK', 'local'),
        'url_strategy' => \SocialDept\AtpParity\Enums\BlobUrlStrategy::Cdn,
    ],
];

Creating Custom Records

Extend the Record base class for custom AT Protocol records:

use SocialDept\AtpParity\Data\Record;
use Carbon\Carbon;

class PostRecord extends Record
{
    public function __construct(
        public readonly string $text,
        public readonly Carbon $createdAt,
        public readonly ?array $facets = null,
    ) {}

    public static function getLexicon(): string
    {
        return 'app.bsky.feed.post';
    }

    public static function fromArray(array $data): static
    {
        return new static(
            text: $data['text'],
            createdAt: Carbon::parse($data['createdAt']),
            facets: $data['facets'] ?? null,
        );
    }
}

The Record class extends atp-schema's Data and implements atp-client's Recordable interface, ensuring full compatibility with the ecosystem.

Requirements

PHP 8.3+
Laravel 11, 12, or 13
socialdept/atp-schema ^0.4
socialdept/atp-cbor ^0.2
socialdept/atp-client ^0.1 || ^0.2 || ^0.3
socialdept/atp-support ^0.3
socialdept/atp-signals ^2.1

Testing

composer test

Point the shipped assertions at each of your mappers. The second one is the important one: your own writes come back as firehose events, so a mapper whose ingest is not a no-op dirties the row, which resyncs, which writes, which produces the next event.

use SocialDept\AtpParity\Testing\AssertsRecordParity;

$this->assertRecordParity(new PostMapper, $savedPost);

Resources

AT Protocol Documentation
Bluesky API Docs
atp-schema - Generated AT Protocol DTOs
atp-client - AT Protocol HTTP client
atp-signals - Firehose event consumer

Support & Contributing

Found a bug or have a feature request? Open an issue.

Want to contribute? Check out the contribution guidelines.

Upgrading

See UPGRADING.md for breaking changes between versions.

1.0 needs two changes. A mapper now imports nothing until it declares accepts(), and SchemaMapper takes an acceptance argument. A mapper that overrides shouldImport() itself is unaffected.

Everything else in 1.0 is additive: declared fields, record upcasting, the blob resolver, and the parity assertions are all opt-in.

Changelog

Please see changelog for recent changes.

Credits

Miguel Batres - founder & lead maintainer
All contributors

License

Parity is open-source software licensed under the MIT license.

Built for the Atmosphere - By Social Dept.

socialdept / atp-parity

Package info

Security

Statistics