view release on metacpan or  search on metacpan

lib/EV/Nats/ObjectStore.pm  view on Meta::CPAN

    my $prefix = '$O.' . $self->{bucket} . '.M.';
    my @try = $prefix . _encode_name($name);
    my $legacy = $prefix . _encode_name_legacy($name);
    push @try, $legacy if $legacy ne $try[0];

    my $attempt;
    $attempt = sub {
        my $meta_subj = shift @try;
        if (!defined $meta_subj) {
            # Break the $attempt self-cycle before firing $cb.
            undef $attempt;
            return $cb->(undef, undef, 0, undef);
        }
        $self->{js}->_json_api(
            'STREAM.MSG.GET.' . $self->{stream},
            { last_by_subj => $meta_subj },
            sub {
                my ($resp, $err) = @_;
                if ($err) {
                    # Clean miss on this subject -> try the fallback form.
                    return $attempt->() if $err =~ /no message found|10037/;
                    undef $attempt;
                    return $cb->(undef, undef, 0, $err);
                }
                my $msg = $resp->{message};
                return $attempt->() if !$msg;
                my $tombstone = EV::Nats::JetStream::msg_is_tombstone($msg);
                require MIME::Base64;
                my $meta_json = MIME::Base64::decode_base64($msg->{data} || '');
                my ($meta, $derr);
                # A tombstone's payload is empty; only parse real data.
                if (length $meta_json) {
                    ($meta, $derr) = EV::Nats::JetStream::decode_json_or_error($meta_json);
                    if ($derr) {
                        undef $attempt;
                        return $cb->(undef, $meta_subj, 0, "invalid metadata: $derr");
                    }
                }
                my $gone = ($tombstone || ($meta && $meta->{deleted})) ? 1 : 0;
                undef $attempt;
                $cb->($meta, $meta_subj, $gone, undef);
            }
        );
    };
    $attempt->();
}

sub _nuid {
    my @chars = ('A'..'Z', 'a'..'z', '0'..'9');
    join '', map { $chars[rand @chars] } 1..22;
}

# padded base64url of a byte/most-strings value, matching Go URLEncoding
sub _b64url {
    my $bytes = shift;
    utf8::encode($bytes) if utf8::is_utf8($bytes);   # names may be wide
    require MIME::Base64;
    (my $e = MIME::Base64::encode_base64($bytes, '')) =~ tr{+/}{-_};
    return $e;                                        # keeps '=' padding
}
# inverse: padded-or-unpadded base64url -> UTF-8-decoded string
sub _b64url_dec {
    my $s = shift;
    $s =~ tr{-_}{+/};
    require MIME::Base64;
    my $b = MIME::Base64::decode_base64($s);          # tolerant of padding
    utf8::decode($b);
    return $b;
}

# ADR-20 name encoding: padded base64url (nats.go). The 0.03-0.05 %XX
# percent form is kept as *_legacy for the dual-read fallback.
sub _encode_name { _b64url($_[0]) }
sub _decode_name { _b64url_dec($_[0]) }
sub _encode_name_legacy { my $n = $_[0]; $n =~ s/([^A-Za-z0-9._-])/sprintf("%%%02X", ord($1))/ge; $n }
sub _decode_name_legacy { my $n = $_[0]; $n =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/ge; $n }

# Best-effort name decode for list(): the base64url form wins when
# re-encoding round-trips, otherwise fall back to the legacy %XX form.
sub _decode_name_best {
    my $tok = shift;
    my $dec = _b64url_dec($tok);
    return $dec if _encode_name($dec) eq $tok;
    return _decode_name_legacy($tok);
}

1;

=head1 NAME

EV::Nats::ObjectStore - Chunked object store on top of NATS JetStream

=head1 SYNOPSIS

    use EV;
    use EV::Nats;
    use EV::Nats::JetStream;
    use EV::Nats::ObjectStore;

    my $nats = EV::Nats->new(host => '127.0.0.1');
    my $js   = EV::Nats::JetStream->new(nats => $nats);
    my $os   = EV::Nats::ObjectStore->new(js => $js, bucket => 'files');

    $os->create_bucket({}, sub {
        $os->put('report.pdf', $pdf_data, sub {
            my ($info, $err) = @_;
            print "stored: $info->{size} bytes in $info->{chunks} chunks\n";
            $os->get('report.pdf', sub {
                my ($data, $err, $meta) = @_;
                print "got $meta->{size} bytes back\n";
            });
        });
    });

    EV::run;

=head1 DESCRIPTION

An object-store bucket is a JetStream stream named C<OBJ_E<lt>bucketE<gt>>
with two subject groups:

lib/EV/Nats/ObjectStore.pm  view on Meta::CPAN

Callback: C<($info, $err)>.

=head2 delete_bucket

    delete_bucket($cb)

Tear down the underlying stream. Callback: C<($info, $err)>.

=head2 put

    put($name, $data, $cb)

Store C<$data> under C<$name>, automatically chunked. Each chunk is
published with JetStream ack; the metadata entry is written last
(with a C<Nats-Rollup: sub> header, so the meta subject holds exactly
one message) so a partial upload doesn't surface a half-stored object.
Overwriting an existing name purges the previous object's chunks after
the new metadata is acknowledged. Callback: C<($info, $err)> where
C<$info> is C<{ name, size, chunks, seq }>.

=head2 get

    get($name, $cb)

Retrieve a previously-stored object. Callback: C<($data, $err, $meta)>.
C<$data> is C<undef> if the object does not exist or has been deleted
(both the old C<KV-Operation> tombstone and the nats.go
C<"deleted":true> marker are recognised). On digest mismatch, C<$data>
is C<undef> and C<$err> is C<"digest mismatch">. Digests written by
0.03/0.04 (hex), 0.05 (unpadded base64url) and 0.06+/nats.go (padded
base64url) all verify.

=head2 delete

    delete($name, [$cb])

Publishes a C<"deleted":true> metadata marker with a C<Nats-Rollup:
sub> header (the nats.go form; the PubAck doubles as the flush fence),
then best-effort purges the object's chunks under
C<$O.E<lt>bucketE<gt>.C.E<lt>nuidE<gt>>. Marking is done first, so a
purge hiccup cannot leave a "deleted" object still readable. Idempotent:
deleting a missing or already-deleted object is a no-op success and
writes no marker. Callback: C<($ok, $err)>.

=head2 info

    info($name, $cb)

Fetch only the metadata entry for an object, without downloading
chunks. Callback: C<(\%meta, $err)>; C<\%meta> is C<undef> if the
object does not exist or was deleted (the deletion marker is
recognised). This is the recommended way to filter live objects out
of a L</list> result.

=head2 list

    list($cb)

List object names in the bucket. Callback: C<(\@names, $err)>.
Names written by 0.03-0.05 (legacy %XX encoding) and by 0.06+/nats.go
(base64url) are both decoded. Deleted entries still appear in the
listing -- filtering them would cost a per-name metadata round-trip,
so call C<info> to filter.

=head2 status

    status($cb)

Returns a snapshot hashref:

    { bucket => $name, bytes => $n, sealed => 0|1 }

C<sealed> reflects the underlying stream's C<config.sealed> flag;
this client never seals on its own, so unless someone manually
sealed the stream out-of-band the value is always 0.

Callback: C<(\%status, $err)>.

=head1 SEE ALSO

L<EV::Nats>, L<EV::Nats::JetStream>, L<EV::Nats::KV>,
L<NATS Object Store|https://docs.nats.io/using-nats/developer/develop_jetstream/object>.

=cut