File Coverage

lib/At.pm

Criterion	Covered	Total	%
statement	59	78	75.6
branch	0	4	0.0
condition			n/a
subroutine	20	24	83.3
pod			n/a
total	79	106	74.5

use v5.42;
use feature 'class';
no warnings 'experimental::class', 'experimental::builtin', 'experimental::for_list';    # Be quiet.

#~ |---------------------------------------|
#~ |------3-33-----------------------------|
#~ |-5-55------4-44-5-55----353--3-33-/1~--|
#~ |---------------------335---33----------|
class At 1.6 {
    use Carp qw[];
    use experimental 'try';
    use File::ShareDir::Tiny qw[dist_dir];
    use JSON::PP             qw[decode_json encode_json];
    use Path::Tiny           qw[path];
    use Digest::SHA          qw[sha256];
    use MIME::Base64         qw[encode_base64url];
    use Crypt::PK::ECC;
    use Crypt::PRNG qw[random_string];
    use Time::Moment;    # Internal; standardize around Zulu
    use URI;
    use warnings::register;
    use At::Error;
    use At::Protocol::URI;
    use At::Protocol::Session;
    use At::UserAgent;
    field $share    : reader : param = ();
    field %lexicons : reader;
    field $http     : reader : param = ();
    field $lexicon_paths_param : param(lexicon_paths) = [];
    field @lexicon_paths;
    field $host      : param : reader //= 'bsky.social';
    field $ipfs_node : param : reader //= undef;
    field $bitswap;
    method set_host ($new) { $host = $new }
    field $session = ();
    field $oauth_state;
    field $dpop_key;
    field %ratelimits : reader = (    # https://docs.bsky.app/docs/advanced-guides/rate-limits
        global        => {},
        updateHandle  => {},          # per DID
        createSession => {},          # per handle
        deleteAccount => {},          # by IP
        resetPassword => {}           # by IP
    );
    ADJUST {
        if ($ipfs_node) {
            require InterPlanetary::Protocol::Bitswap;
            $bitswap = InterPlanetary::Protocol::Bitswap->new( node => $ipfs_node );
        }
        if ( !defined $share ) {
            try { $share = dist_dir('At') }
            catch ($e) { $share = 'share' }
        }
        $share = path($share) unless builtin::blessed $share;
        @lexicon_paths = map { path($_) } ( ref $lexicon_paths_param eq 'ARRAY' ? @$lexicon_paths_param : ($lexicon_paths_param) );
        if ( !defined $http ) {
            my $ua_class;
            try {
                require Mojo::UserAgent;
                $ua_class = 'At::UserAgent::Mojo';
            }
            catch ($e) {
                $ua_class = 'At::UserAgent::Tiny';
            }
            $http = $ua_class->new();
        }
        $host = 'https://' . $host unless $host =~ /^https?:/;
        $host = URI->new($host)    unless builtin::blessed $host;
    }

    # OAuth Implementation
    method _get_dpop_key() {
        unless ($dpop_key) {
            $dpop_key = Crypt::PK::ECC->new();
            $dpop_key->generate_key('secp256r1');
        }
        return $dpop_key;
    }

    method oauth_discover ($handle) {
        my $res = $self->resolve_handle($handle);
        if ( builtin::blessed($res) && $res->isa('At::Error') ) { $res->throw; }
        return unless $res && $res->{did};
        my $pds = $self->pds_for_did( $res->{did} );
        unless ($pds) { die 'Could not resolve PDS for DID: ' . $res->{did}; }
        my ($protected) = $http->get( $pds . '/.well-known/oauth-protected-resource' );
        if ( builtin::blessed($protected) && $protected->isa('At::Error') ) { $protected->throw; }
        return unless $protected && $protected->{authorization_servers};
        my $auth_server = $protected->{authorization_servers}[0];
        my ($metadata) = $http->get( $auth_server . '/.well-known/oauth-authorization-server' );
        if ( builtin::blessed($metadata) && $metadata->isa('At::Error') ) { $metadata->throw; }
        return { pds => $pds, auth_server => $auth_server, metadata => $metadata, did => $res->{did} };
    }

    method oauth_start ( $handle, $client_id, $redirect_uri, $scope = 'atproto' ) {
        my $discovery = $self->oauth_discover($handle);
        die 'Failed to discover OAuth metadata for ' . $handle unless $discovery;
        my $chars          = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-._~';
        my $code_verifier  = Crypt::PRNG::random_string_from( $chars, 43 );
        my $code_challenge = encode_base64url( sha256($code_verifier) );
        $code_challenge =~ s/=+$//;
        my $state = Crypt::PRNG::random_string_from( $chars, 16 );
        $oauth_state = {
            discovery     => $discovery,
            code_verifier => $code_verifier,
            state         => $state,
            redirect_uri  => $redirect_uri,
            client_id     => $client_id,
            handle        => $handle,
            scope         => $scope
        };

        # Prepare UA for DPoP
        $http->set_tokens( undef, undef, 'DPoP', $self->_get_dpop_key() );
        my $par_endpoint = $discovery->{metadata}{pushed_authorization_request_endpoint};
        my $par_content  = {
            client_id             => $client_id,
            response_type         => 'code',
            code_challenge        => $code_challenge,
            code_challenge_method => 'S256',
            redirect_uri          => $redirect_uri,
            state                 => $state,
            scope                 => $scope,
            aud                   => $discovery->{pds},
        };
        say '[DEBUG] [At] PAR request: ' . JSON::PP->new->ascii->encode($par_content) if $ENV{DEBUG};
        my ($par_res) = $http->post(
            $par_endpoint => {
                headers  => { DPoP => $http->_generate_dpop_proof( $par_endpoint, 'POST', 1 ) },
                encoding => 'form',
                content  => $par_content,
                skip_ath => 1
            }
        );
        die 'PAR failed: ' . ( $par_res . '' )                                     if builtin::blessed $par_res;
        say '[DEBUG] [At] PAR response: ' . JSON::PP->new->ascii->encode($par_res) if $ENV{DEBUG};
        my $auth_uri = URI->new( $discovery->{metadata}{authorization_endpoint} );
        $auth_uri->query_form( client_id => $client_id, request_uri => $par_res->{request_uri} );
        return $auth_uri->as_string;
    }

    method oauth_callback ( $code, $state ) {
        die 'OAuth state mismatch' unless $oauth_state && $state eq $oauth_state->{state};
        my $token_endpoint = $oauth_state->{discovery}{metadata}{token_endpoint};
        my $key            = $self->_get_dpop_key();
        my ($token_res)    = $http->post(
            $token_endpoint => {
                headers  => { DPoP => $http->_generate_dpop_proof( $token_endpoint, 'POST', 1 ) },
                encoding => 'form',
                content  => {
                    grant_type    => 'authorization_code',
                    code          => $code,
                    client_id     => $oauth_state->{client_id},
                    redirect_uri  => $oauth_state->{redirect_uri},
                    code_verifier => $oauth_state->{code_verifier},
                    aud           => $oauth_state->{discovery}{pds}
                },
                skip_ath => 1
            }
        );
        die 'Token exchange failed: ' . ( $token_res . '' )                            if builtin::blessed $token_res;
        say '[DEBUG] [At] Token response: ' . JSON::PP->new->ascii->encode($token_res) if $ENV{DEBUG};
        $session = At::Protocol::Session->new(
            did          => $token_res->{sub},
            accessJwt    => $token_res->{access_token},
            refreshJwt   => $token_res->{refresh_token},
            handle       => $oauth_state->{handle},
            token_type   => 'DPoP',
            dpop_key_jwk => $key->export_key_jwk('private'),
            client_id    => $oauth_state->{client_id},
            scope        => $token_res->{scope},
            pds          => $oauth_state->{discovery}{pds}
        );
        $self->set_host( $oauth_state->{discovery}{pds} );
        $http->set_tokens( $token_res->{access_token}, $token_res->{refresh_token}, 'DPoP', $key );
    }

    method oauth_refresh() {
        return unless $session && $session->refreshJwt && $session->token_type eq 'DPoP';
        my $discovery = $self->oauth_discover( $session->handle );
        return unless $discovery;
        my $token_endpoint  = $discovery->{metadata}{token_endpoint};
        my $key             = $self->_get_dpop_key();
        my $refresh_content = {
            grant_type    => 'refresh_token',
            refresh_token => $session->refreshJwt,
            client_id     => $session->client_id // '',
            aud           => $discovery->{pds},
        };
        say '[DEBUG] [At] Refresh request: ' . JSON::PP->new->ascii->encode($refresh_content) if $ENV{DEBUG};
        my ($token_res) = $http->post(
            $token_endpoint => {
                headers  => { DPoP => $http->_generate_dpop_proof( $token_endpoint, 'POST', 1 ) },
                encoding => 'form',
                content  => $refresh_content,
                skip_ath => 1
            }
        );
        die 'Refresh failed: ' . ( $token_res . '' ) if builtin::blessed $token_res;
        $session = At::Protocol::Session->new(
            did          => $token_res->{sub},
            accessJwt    => $token_res->{access_token},
            refreshJwt   => $token_res->{refresh_token},
            handle       => $session->handle,
            token_type   => 'DPoP',
            dpop_key_jwk => $key->export_key_jwk('private'),
            client_id    => $session->client_id,
            scope        => $token_res->{scope},
            pds          => $discovery->{pds}
        );
        $self->set_host( $discovery->{pds} );
        return $http->set_tokens( $token_res->{access_token}, $token_res->{refresh_token}, 'DPoP', $key );
    }

    method collection_scope ( $collection, $action = 'create' ) {
        return "repo:$collection?action=$action";
    }

    # Legacy Auth
    method login( $identifier, $password ) {
        warnings::warnif( At => 'login() (com.atproto.server.createSession) is deprecated. Please use OAuth instead.' );
        my $res = $self->post( 'com.atproto.server.createSession' => { identifier => $identifier, password => $password } );
        if   ( $res && !builtin::blessed($res) ) { $session = At::Protocol::Session->new(%$res); }
        else                                     { $session = $res; }
        return $session ? $http->set_tokens( $session->accessJwt, $session->refreshJwt, undef, undef ) : $session;
    }

    method resume ( $accessJwt, $refreshJwt, $token_type = 'Bearer', $dpop_key_jwk = (), $client_id = (), $handle = (), $pds = (), $scope = () ) {
        my $access  = $self->_decode_token($accessJwt);
        my $refresh = $self->_decode_token($refreshJwt);
        return unless $access;
        my $key;
        if ( $token_type eq 'DPoP' && $dpop_key_jwk ) {
            $key = Crypt::PK::ECC->new();
            $key->import_key( \$dpop_key_jwk );
            $dpop_key = $key;
        }
        if ( $refresh && time > $access->{exp} && time < $refresh->{exp} ) {
            if ( $token_type eq 'DPoP' ) { return $self->oauth_refresh(); }
            else {
                my $res = $self->post( 'com.atproto.server.refreshSession' => { refreshJwt => $refreshJwt } );
                if   ( $res && !builtin::blessed($res) ) { $session = At::Protocol::Session->new(%$res); }
                else                                     { $session = $res; }
                return $session ? $http->set_tokens( $session->accessJwt, $session->refreshJwt, $token_type, $key ) : $session;
            }
        }
        $session = At::Protocol::Session->new(
            did          => $access->{sub},
            accessJwt    => $accessJwt,
            refreshJwt   => $refreshJwt,
            token_type   => $token_type,
            dpop_key_jwk => $dpop_key_jwk,
            client_id    => $client_id,
            handle       => $handle,
            pds          => $pds,
            scope        => $scope
        );
        $self->set_host($pds) if $pds;
        return $http->set_tokens( $accessJwt, $refreshJwt, $token_type, $key );
    }

    method _decode_token ($token) {
        return unless defined $token;
        use MIME::Base64 qw[decode_base64];
        my ( $header, $payload, $sig ) = split /\./, $token;
        return unless defined $payload;
        $payload =~ tr[-_][+/];
        try {
            return decode_json decode_base64 $payload;
        }
        catch ($e) {
            return;
        }
    }

    # XRPC & Lexicons
    method _locate_lexicon($fqdn) {
        $fqdn // return undef;
        $fqdn =~ s/^#//;    # Strip leading hash
        unless ( defined $lexicons{$fqdn} ) {
            my $base_fqdn = $fqdn =~ s[#(.+)$][]r;
            return undef unless $base_fqdn;
            my @namespace = split /\./, $base_fqdn;
            return undef unless @namespace;
            my @search = (
                @lexicon_paths,
                $share->child('lexicons'),
                defined $ENV{HOME} ? path( $ENV{HOME}, '.cache', 'atproto', 'lexicons' ) : (),
                path( 'share', 'lexicons' )
            );
            my $lex_file;
            for my $dir (@search) {
                next unless defined $dir;
                my $possible = $dir->child( @namespace[ 0 .. $#namespace - 1 ], $namespace[-1] . '.json' );
                if ( $possible->exists ) { $lex_file = $possible; last; }
            }
            if ( !$lex_file ) { $lex_file = $self->_fetch_lexicon($base_fqdn); }
            if ( $lex_file && $lex_file->exists ) {
                my $json = decode_json $lex_file->slurp_raw;
                for my $def ( keys %{ $json->{defs} } ) {
                    $lexicons{ $base_fqdn . ( $def eq 'main' ? '' : '#' . $def ) } = $json->{defs}{$def};
                    $lexicons{ $base_fqdn . '#main' } = $json->{defs}{$def} if $def eq 'main';
                }
            }
        }
        $lexicons{$fqdn};
    }

    method _fetch_lexicon($base_fqdn) {
        my @namespace = split /\./, $base_fqdn;
        my $rel_path  = join( '/', @namespace[ 0 .. $#namespace - 1 ], $namespace[-1] . '.json' );
        my $url       = 'https://raw.githubusercontent.com/bluesky-social/atproto/main/lexicons/' . $rel_path;
        my ( $content, $headers ) = $http->get($url);
        if ( $content && !builtin::blessed($content) ) {
            my $cache_dir = defined $ENV{HOME} ? path( $ENV{HOME}, '.cache', 'atproto', 'lexicons' ) : path( '.cache', 'atproto', 'lexicons' );
            $cache_dir->mkpath;
            my $lex_file = $cache_dir->child( @namespace[ 0 .. $#namespace - 1 ], $namespace[-1] . '.json' );
            $lex_file->parent->mkpath;
            $lex_file->spew_raw( builtin::blessed($content) ? encode_json($content) : $content );
            return $lex_file;
        }
        return;
    }

    method get( $fqdn, $args = (), $headers = {} ) {
        my $lexicon  = $self->_locate_lexicon($fqdn);
        my $category = $fqdn =~ /^com\.atproto\.repo\./ ? 'repo'                          : 'global';
        my $meta     = $category eq 'repo'              ? ( $args->{repo} // $self->did ) : ();
        $self->_ratecheck( $category, $meta );
        my ( $content, $res_headers )
            = $http->get( sprintf( '%s/xrpc/%s', $host, $fqdn ), { defined $args ? ( content => $args ) : (), headers => $headers } );
        $self->ratelimit_( $res_headers, $category, $meta );
        if ( $lexicon && !builtin::blessed $content ) {
            $content = $self->_coerce( $fqdn, $lexicon->{output}{schema}, $content );
        }
        wantarray ? ( $content, $res_headers ) : $content;
    }

    method post( $fqdn, $args = (), $headers = {} ) {
        my @namespace = split /\./, $fqdn;
        my $lexicon   = $self->_locate_lexicon($fqdn);

        # Categorize according to bsky specs
        my $category = 'global';
        my $meta     = ();
        if ( $fqdn =~ /^com\.atproto\.server\.createSession$/ ) {
            $category = 'auth';
            $meta     = $args->{identifier};
        }
        elsif ( $fqdn =~ /^com\.atproto\.repo\./ ) {
            $category = 'repo';
            $meta     = $args->{repo} // $self->did;
        }
        elsif ( $namespace[-1] =~ m[^(updateHandle|createAccount|deleteAccount|resetPassword)$] ) {
            $category = $namespace[-1];
            $meta     = $args->{did} // $args->{handle} // $args->{email};
        }
        $self->_ratecheck( $category, $meta );
        my ( $content, $res_headers )
            = $http->post( sprintf( '%s/xrpc/%s', $host, $fqdn ), { defined $args ? ( content => $args ) : (), headers => $headers } );
        $self->ratelimit_( $res_headers, $category, $meta );
        if ( $lexicon && !builtin::blessed $content ) {
            $content = $self->_coerce( $fqdn, $lexicon->{output}{schema}, $content );
        }
        return wantarray ? ( $content, $res_headers ) : $content;
    }
    method subscribe( $id, $cb ) { $self->http->websocket( sprintf( '%s/xrpc/%s', $host, $id ), $cb ); }

    method firehose ( $callback, $url = () ) {
        require At::Protocol::Firehose;
        return At::Protocol::Firehose->new( at => $self, callback => $callback, defined $url ? ( url => $url ) : () );
    }

    # Coercion Logic
    my %coercions = (
        array => method( $namespace, $schema, $data ) {
            [ map { $self->_coerce( $namespace, $schema->{items}, $_ ) } @$data ]
        },
        boolean => method( $namespace, $schema, $data ) { !!$data },
        bytes   => method( $namespace, $schema, $data ) {$data},
        blob    => method( $namespace, $schema, $data ) {$data},
        integer => method( $namespace, $schema, $data ) { int $data },
        object  => method( $namespace, $schema, $data ) {
            for my ( $name, $subschema )( %{ $schema->{properties} } ) {
                $data->{$name} = $self->_coerce( $namespace, $subschema, $data->{$name} );
            }
            $data;
        },
        ref => method( $namespace, $schema, $data ) {
            my $target_namespace = $self->_resolve_namespace( $namespace, $schema->{ref} );
            my $lexicon          = $self->_locate_lexicon($target_namespace);
            return $data unless $lexicon;
            $self->_coerce( $target_namespace, $lexicon, $data );
        },
        union   => method( $namespace, $schema, $data ) {$data},
        unknown => method( $namespace, $schema, $data ) {$data},
        string  => method( $namespace, $schema, $data ) {
            $data // return ();
            if ( defined $schema->{format} ) {
                if    ( $schema->{format} eq 'uri' )    { return URI->new($data); }
                elsif ( $schema->{format} eq 'at-uri' ) { return At::Protocol::URI->new($data); }
                elsif ( $schema->{format} eq 'datetime' ) {
                    return $data =~ /\D/ ? Time::Moment->from_string($data) : Time::Moment->from_epoch($data);
                }
                elsif ( $schema->{format} eq 'did' ) {
                    require At::Protocol::DID;
                    return At::Protocol::DID->new($data);
                }
                elsif ( $schema->{format} eq 'handle' ) {
                    require At::Protocol::Handle;
                    try { return At::Protocol::Handle->new($data); }
                    catch ($e) { return $data; }
                }
            }
            $data;
        }
    );

    method _coerce ( $namespace, $schema, $data ) {
        $data // return ();
        return $coercions{ $schema->{type} }->( $self, $namespace, $schema, $data ) if defined $coercions{ $schema->{type} };
        return $data;
    }

    method _resolve_namespace ( $l, $r ) {
        return $r if $r =~ m[\.];    # Absolute (has dots)
        my $base = $l =~ s[#(.+)$][]r;
        return $base . '#' . $r;     # Relative to base of current FQDN
    }

    # Identity & Helpers
    method did()                   { $session ? $session->did . '' : undef; }
    method resolve_handle($handle) { $self->get( 'com.atproto.identity.resolveHandle' => { handle => $handle } ); }

    method resolve_did_to_handle ($did) {
        my $doc = $self->resolve_did($did);
        return $doc->{alsoKnownAs}[0] =~ s/^at:\/\///r if $doc && $doc->{alsoKnownAs};
        return;
    }

    method upload_blob ( $data, $mime_type ) {
        $self->post( 'com.atproto.repo.uploadBlob' => { content => $data, headers => { 'Content-Type' => $mime_type } } );
    }

    method create_record ( $collection, $record, $rkey = undef ) {
        $self->post( 'com.atproto.repo.createRecord' =>
                { repo => $self->did, collection => $collection, record => $record, defined $rkey ? ( rkey => $rkey ) : () } );
    }

    method delete_record ( $collection, $rkey ) {
        $self->post( 'com.atproto.repo.deleteRecord' => { repo => $self->did, collection => $collection, rkey => $rkey } );
    }

    method put_record ( $collection, $rkey, $record, $swapRecord = undef ) {
        $self->post(
            'com.atproto.repo.putRecord' => {
                repo       => $self->did,
                collection => $collection,
                rkey       => $rkey,
                record     => $record,
                defined $swapRecord ? ( swapRecord => $swapRecord ) : ()
            }
        );
    }

    method apply_writes ( $writes, $swapCommit = undef ) {
        $self->post(
            'com.atproto.repo.applyWrites' => { repo => $self->did, writes => $writes, defined $swapCommit ? ( swapCommit => $swapCommit ) : () } );
    }

    method resolve_did ($did) {
        if ( $did =~ /^did:plc:(.+)$/ ) {
            my ($content) = $http->get( 'https://plc.directory/' . $did );
            return $content;
        }
        elsif ( $did =~ /^did:web:(.+)$/ ) {
            my $domain = $1;
            $domain =~ s/:/\//g;
            my ($content) = $http->get("https://$domain/.well-known/did.json");
            return $content;
        }
        return;
    }

    method pds_for_did ($did) {
        my $doc = $self->resolve_did($did);
        return unless $doc && ref $doc eq 'HASH' && $doc->{service};
        for my $service ( @{ $doc->{service} } ) {
            return $service->{serviceEndpoint} if $service->{type} eq 'AtprotoPersonalDataServer';
        }
        return;
    }

    method peer_id_for_did ($did) {
        my $doc = $self->resolve_did($did);
        return unless $doc && ref $doc eq 'HASH' && $doc->{verificationMethod};

        # Look for the primary signing key (usually the first one)
        my $vm                = $doc->{verificationMethod}[0];
        my $pub_key_multibase = $vm->{publicKeyMultibase} // return;

        # publicKeyMultibase for secp256k1 in atproto usually starts with 'z' (base58btc)
        # and has a multicodec prefix.
        require InterPlanetary::Multibase;
        my $raw = InterPlanetary::Multibase->decode($pub_key_multibase);

        # For secp256k1 (0xe7 multicodec), we need to extract the 33-byte compressed key
        # and wrap it in the libp2p Protobuf PublicKey.
        require Net::Libp2p::Crypto;
        require InterPlanetary::Utils;

        # Simplified: assume it's secp256k1 for now as per bsky standard
        my $key_bytes = substr( $raw, 2 );    # Skip multicodec prefix (usually 2 bytes for 0xe7)

        # Wrap in libp2p Protobuf (Type 2 = Secp256k1)
        my $pk_pb = pack( 'C', ( 1 << 3 ) | 0 ) . InterPlanetary::Utils::encode_varint(2);
        $pk_pb .= pack( 'C', ( 2 << 3 ) | 2 ) . InterPlanetary::Utils::encode_varint( length($key_bytes) ) . $key_bytes;
        return Net::Libp2p::Crypto->peer_id_from_public_key($pk_pb);
    }
    method session() { $session //= $self->get('com.atproto.server.getSession'); $session; }

    method get_block ( $cid_str, $target_peer_id = undef, $did = undef ) {
        if ($ipfs_node) {
            my $cid  = InterPlanetary::CID->from_string($cid_str);
            my $data = $ipfs_node->blockstore->get($cid);
            return Future->done($data) if $data;
            if ($target_peer_id) {

                #~ say "[IPFS] Block $cid_str not found locally, attempting Bitswap from $target_peer_id...";
                return $ipfs_node->host->dial( $target_peer_id, '/ipfs/bitswap/1.2.0' )->then(
                    sub ($ss) {

                        # The local bitswap handler needs to 'own' this stream to read the response
                        $bitswap->handle_stream($ss);
                        return $bitswap->request_block( $ss, $cid );
                    }
                )->else(
                    sub {
                        my ($e) = @_;

                        #~ say "[IPFS] Bitswap failed: $e. Falling back to HTTP...";
                        return $self->_get_block_http( $cid_str, $did );
                    }
                );
            }
        }
        return $self->_get_block_http( $cid_str, $did );
    }

    method _get_block_http ( $cid_str, $did ) {

        # If no DID provided, we can't fallback to sync endpoints
        return Future->done(undef) unless $did;

        #~ say "[HTTP] Fetching block $cid_str for $did via com.atproto.sync.getBlocks...";
        # com.atproto.sync.getBlocks returns a CAR file
        return Future->call(
            sub {
                my $car_data = $self->get( 'com.atproto.sync.getBlocks' => { did => $did, cids => [$cid_str] } );
                return undef unless $car_data;
                require Archive::CAR;
                require Archive::CAR::v1;
                my $car = Archive::CAR::v1->new();
                open my $fh, '<', \$car_data;
                $car->read($fh);

                # Archive::CAR blocks is an arrayref of { cid => CIDobj, data => ... }
                for my $block ( @{ $car->blocks } ) {

                    # CIDs are complex objects, to_string should work for matching
                    # But the CID in the CAR might be different multibase than $cid_str
                    # For now, simplistic comparison or raw check
                    return $block->{data} if $block->{cid}->to_string eq $cid_str;
                }
                return undef;
            }
        );
    }
    method get_repo_head ($did) { $self->get( 'com.atproto.sync.getHead' => { did => $did } ) }
    sub _now                    { Time::Moment->now }
    method _duration ($seconds) { $seconds || return '0 seconds'; $seconds = abs $seconds; return $seconds . ' seconds' }

    method ratelimit_ ( $headers, $type, $meta //= () ) {
        my %h = map { lc($_) => $headers->{$_} } keys %$headers;
        return unless exists $h{'ratelimit-limit'};
        my $rate = {
            limit     => $h{'ratelimit-limit'},
            remaining => $h{'ratelimit-remaining'},
            reset     => $h{'ratelimit-reset'},
            policy    => $h{'ratelimit-policy'},
        };
        defined $meta ? $ratelimits{$type}{$meta} = $rate : $ratelimits{$type} = $rate;
    }

    method _ratecheck( $type, $meta //= () ) {
        my $rate = defined $meta ? $ratelimits{$type}{$meta} : $ratelimits{$type};
        return unless $rate && $rate->{reset};
        if ( $rate->{remaining} <= 0 && time < $rate->{reset} ) {
            my $wait = $rate->{reset} - time;
            warnings::warnif( At => "Rate limit exceeded for $type. Reset in $wait seconds." );
        }
        elsif ( $rate->{remaining} < ( $rate->{limit} * 0.1 ) ) {
            warnings::warnif( At => "Approaching rate limit for $type ($rate->{remaining} remaining)." );
        }
    }
}
1;
__END__