feat: add JetStream KV (Key-Value Store) support

META6.json

@@ -1,7 +1,7 @@
 {
   "auth": "zef:FCO",
   "authors": [
-    "Fernando Corrêa de Oliveira"
+    "Fernando Corr\u00eaa de Oliveira"
   ],
   "build-depends": [
     "Test::Mock",
@@ -11,7 +11,7 @@
     "JSON::Fast",
     "URL"
   ],
-  "description": "NATS client for Raku with JetStream support",
+  "description": "NATS client for Raku with JetStream and KV support",
   "license": "Artistic-2.0",
   "name": "Nats",
   "perl": "6.d",
@@ -27,17 +27,17 @@
     "Nats::Message": "lib/Nats/Message.rakumod",
     "Nats::Replyable": "lib/Nats/Replyable.rakumod",
     "Nats::Subscription": "lib/Nats/Subscription.rakumod",
-    "Nats::Subscriptions": "lib/Nats/Subscriptions.rakumod"
+    "Nats::Subscriptions": "lib/Nats/Subscriptions.rakumod",
+    "Nats::KV": "lib/Nats/KV.rakumod"
   },
-  "resources": [
-  ],
+  "resources": [],
   "source-url": "https://github.com/FCO/nats.raku.git",
   "tags": [
     "nats",
     "jetstream",
-    "messaging"
-  ],
-  "test-depends": [
+    "messaging",
+    "kv"
   ],
+  "test-depends": [],
   "version": "0.1.0"
 }

lib/Nats.rakumod

@@ -8,6 +8,7 @@ use Nats::Data;
 use Nats::Message;
 use Nats::Subscription;
 use Nats::JetStream;
+use Nats::KV;
 
 has $.socket-class       = IO::Socket::Async;
 has %!subs;
@@ -236,6 +237,10 @@ method stream($name, *@subjects, |c) {
     Nats::Stream.new: :nats(self), :$name, |(:@subjects if @subjects), |c
 }
 
+method kv(Str $bucket, |c) {
+    Nats::KV.new: :nats(self), :$bucket, |c
+}
+
 method !in(|c) {
     self!debug(">>", |c)
 }
@@ -383,6 +388,14 @@ my $stream = $nats.stream: 'mystream', :subjects['foo.>'];
 
 Creates a C<Nats::Stream> object for JetStream operations.
 
+=head2 kv
+
+=begin code :lang<raku>
+my $kv = $nats.kv: 'mybucket';
+=end code
+
+Creates a C<Nats::KV> object for JetStream Key-Value operations.
+
 =head1 JetStream
 
 See C<Nats::JetStream> for stream and consumer management.
@@ -401,6 +414,30 @@ react whenever $consumer.msgs(:batch, :no-wait) {
 }
 =end code
 
+=head1 Key-Value Store (JetStream KV)
+
+See C<Nats::KV> for key-value operations built on JetStream.
+
+=begin code :lang<raku>
+use Nats::KV;
+
+my $kv = $nats.kv: 'mybucket';
+await $kv.create;
+
+$kv.put: 'foo', 'bar';
+say $kv.get: 'foo';          # bar
+
+$kv.delete: 'foo';
+
+# List all keys
+.say for $kv.keys;
+
+# Watch for changes
+react whenever $kv.watch.supply -> $msg {
+    say \"Key changed: { $msg.subject }\";
+}
+=end code
+
 =head1 AUTHOR
 
 Fernando Corrêa de Oliveira <fco@cpan.org>

lib/Nats/KV.rakumod

@@ -0,0 +1,115 @@
+use JSON::Fast;
+
+# ═════════════════════════════════════════════
+# NATS JetStream Key-Value Store
+# ═════════════════════════════════════════════
+#
+# Built on top of JetStream streams. Each KV bucket is a stream
+# with max_msgs_per_subject=1 (last-write-wins) and discard=new.
+#
+# API subjects: $KV.<bucket>.<key>
+#
+# Usage:
+#   my $kv = $nats.kv('mybucket');
+#   await $kv.create;
+#   $kv.put('foo', 'bar');
+#   say $kv.get('foo');  # bar
+#   $kv.delete('foo');
+
+class Nats::KV {
+    has $.nats is required;
+    has Str $.bucket is required;
+    has Str $.description;
+    has Int $.max-age = 0;        # 0 = no TTL
+    has $.stream;
+
+    # Full subject prefix for this bucket
+    method prefix { "\$KV.{ $!bucket }" }
+
+    # Build a stream object configured as a KV bucket
+    method !build-stream {
+        Nats::Stream.new:
+            :$!nats,
+            :name("KV_{ $!bucket }"),
+            :subjects(["{ self.prefix }.>"]),
+            :retention<limits>,
+            :discard<new>,
+            :max-msgs-per-subject(1),
+            :allow-direct,
+            :max-age($!max-age),
+            |($!description ?? :$!description !! Empty),
+    }
+
+    # Create the KV bucket (idempotent — existing buckets are reused)
+    method create {
+        $!stream = self!build-stream;
+        $!stream.create;
+    }
+
+    # Get stream info
+    method info {
+        $!stream.info;
+    }
+
+    # Delete the entire bucket
+    method destroy {
+        $!stream.delete;
+    }
+
+    # ── CRUD operations ──
+
+    # Put a value for a key
+    method put(Str $key, Str() $value) {
+        $!nats.publish: "{ self.prefix }.{ $key }", $value;
+    }
+
+    # Get the current value for a key (returns Str or Nil)
+    method get(Str $key --> Str) {
+        my $resp = $!stream.get-last-msg("{ self.prefix }.{ $key }");
+        return Nil without $resp;
+        my $msg = await $resp.Promise;
+        return Nil unless $msg && $msg.payload && $msg.payload.chars > 0;
+        $msg.payload;
+    }
+
+    # Delete a key (publishes a tombstone — empty payload)
+    method delete(Str $key) {
+        $!nats.publish: "{ self.prefix }.{ $key }", "";
+    }
+
+    # ── Bulk operations ──
+
+    # List all keys in the bucket
+    method keys(--> Seq) {
+        my $resp = $!stream.info;
+        return ().Seq without $resp;
+        my $msg = await $resp.Promise;
+        return ().Seq unless $msg && $msg.payload;
+        my %info = try from-json($msg.payload);
+        return ().Seq if $!;
+        my $prefix = "{ self.prefix }.";
+        my $len    = $prefix.chars;
+        gather {
+            for %info<state><subjects>.List -> $subject {
+                next unless $subject.starts-with($prefix);
+                take $subject.substr($len);
+            }
+        }
+    }
+
+    # ── Watcher ──
+
+    # Subscribe to all changes in the bucket.
+    # Returns a Nats::Subscription. Use $sub.supply to react.
+    method watch {
+        $!nats.subscribe: "{ self.prefix }.>";
+    }
+
+    # ── History ──
+
+    # Get the history of a key (requires the stream to have max_msgs_per_subject > 1).
+    # Not supported in this implementation (KV buckets use max_msgs_per_subject=1).
+    method history(Str $key) {
+        die "history() requires max-msgs-per-subject > 1. This KV bucket uses the default (1).";
+    }
+}

t/kv.rakutest

@@ -0,0 +1,86 @@
+use Test;
+use Nats;
+use Nats::KV;
+
+plan 9;
+
+my $nats = Nats.new: :servers[%*ENV<NATS_URL> // 'nats://localhost:4222'];
+await $nats.start;
+$nats.connect;
+
+my $bucket = 'test-kv-' ~ (^10000).pick;
+
+subtest 'create bucket', {
+    my $kv = $nats.kv($bucket);
+    my $resp = $kv.create;
+    my $msg = await $resp.Promise;
+    ok $msg, 'bucket created';
+    ok $msg.payload, 'response has payload';
+}
+
+subtest 'put and get', {
+    my $kv = $nats.kv($bucket);
+    $kv.put('hello', 'world');
+    sleep 0.1;  # give NATS time to process
+    my $val = $kv.get('hello');
+    is $val, 'world', 'get returns put value';
+}
+
+subtest 'get missing key', {
+    my $kv = $nats.kv($bucket);
+    my $val = $kv.get('nonexistent');
+    ok !$val.defined, 'get missing key returns Nil';
+}
+
+subtest 'overwrite value', {
+    my $kv = $nats.kv($bucket);
+    $kv.put('counter', '1');
+    $kv.put('counter', '2');
+    sleep 0.1;
+    is $kv.get('counter'), '2', 'last write wins';
+}
+
+subtest 'delete key', {
+    my $kv = $nats.kv($bucket);
+    $kv.put('temp', 'data');
+    sleep 0.1;
+    $kv.delete('temp');
+    sleep 0.1;
+    my $val = $kv.get('temp');
+    ok !$val.defined || $val eq '', 'delete removes value';
+}
+
+subtest 'list keys', {
+    my $kv = $nats.kv($bucket);
+    $kv.put('alpha', '1');
+    $kv.put('beta',  '2');
+    $kv.put('gamma', '3');
+    sleep 0.2;
+    my @keys = $kv.keys;
+    ok @keys.elems >= 3, "has at least 3 keys (got {@keys.elems})";
+    ok @keys.grep('alpha'), 'contains alpha';
+    ok @keys.grep('beta'),  'contains beta';
+}
+
+subtest 'watch changes', {
+    my $kv = $nats.kv($bucket);
+    my $sub = $kv.watch;
+    my $p = $sub.supply.head.Promise;
+    $kv.put('watched', 'changed');
+    await Promise.anyof: $p, Promise.in(2);
+    ok $p.so, 'watch received change';
+    $nats.unsubscribe: $sub.sid;
+}
+
+subtest 'kv via Nats.kv method', {
+    my $kv = $nats.kv($bucket);
+    ok $kv ~~ Nats::KV, 'Nats.kv returns Nats::KV';
+}
+
+subtest 'destroy bucket', {
+    my $kv = $nats.kv($bucket);
+    my $resp = $kv.destroy;
+    ok $resp, 'bucket destroyed';
+}
+
+done-testing;