fix: use prefix field instead of regex for subscription prefix filters (#378)

The addPrefix methods in SubscriptionFilter were incorrectly calling
builder.setRegex() instead of builder.addPrefix(), causing prefix filters
to be sent as regex matches on the wire. This could match streams that
contain the prefix anywhere in the name rather than only at the start.

Fixes #374

Co-authored-by: William Chong <william-chong@outlook.com>

Author: github-actions[bot]

src/main/java/io/kurrent/dbclient/SubscriptionFilter.java

@@ -81,7 +81,7 @@ Persistent.CreateReq.AllOptions.FilterOptions.Expression.Builder setRegEx(String
         }
 
         void addPrefix(Persistent.CreateReq.AllOptions.FilterOptions.Expression.Builder builder, String prefix) {
-            builder.setRegex(prefix);
+            builder.addPrefix(prefix);
         }
 
         void setNoFilter() {
@@ -133,7 +133,7 @@ StreamsOuterClass.ReadReq.Options.FilterOptions.Expression.Builder setRegEx(Stri
         }
 
         void addPrefix(StreamsOuterClass.ReadReq.Options.FilterOptions.Expression.Builder builder, String prefix) {
-            builder.setRegex(prefix);
+            builder.addPrefix(prefix);
         }
 
         void setNoFilter() {

src/test/java/io/kurrent/dbclient/streams/SubscriptionTests.java

@@ -139,6 +139,100 @@ public void onCancelled(Subscription subscription, Throwable throwable) {
         subscription.stop();
     }
 
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    default void testSubscribeToAllByStreamPrefix() throws Throwable {
+        String streamPrefix = generateName();
+        String matchingStream = streamPrefix + "-matching";
+        String eventType = generateName();
+        int expectedCount = 5;
+
+        final CountDownLatch receivedEvents = new CountDownLatch(expectedCount);
+        final AtomicInteger count = new AtomicInteger(0);
+
+        Subscription subscription = getDefaultClient().subscribeToAll(new SubscriptionListener() {
+            @Override
+            public void onEvent(Subscription subscription, ResolvedEvent event) {
+                count.incrementAndGet();
+                receivedEvents.countDown();
+            }
+
+            @Override
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null)
+                    return;
+
+                Assertions.fail(throwable);
+            }
+        }, SubscribeToAllOptions.get()
+                .filter(SubscriptionFilter
+                        .newBuilder()
+                        .addStreamNamePrefix(streamPrefix)
+                        .build())).get();
+
+        getDefaultClient().appendToStream(matchingStream, generateEvents(expectedCount, eventType).iterator()).get();
+
+        receivedEvents.await();
+        Assertions.assertEquals(expectedCount, count.get());
+        subscription.stop();
+    }
+
+    @Test
+    @Timeout(value = 2, unit = TimeUnit.MINUTES)
+    default void testSubscribeToAllByStreamPrefixExcludesNonPrefix() throws Throwable {
+        String streamPrefix = generateName();
+        String matchingStream = streamPrefix + "-matching";
+        String nonMatchingStream = "not-" + streamPrefix;
+        String eventType = generateName();
+        int expectedCount = 5;
+
+        final CountDownLatch receivedEvents = new CountDownLatch(expectedCount);
+        final AtomicInteger count = new AtomicInteger(0);
+        final ArrayList<String> receivedStreamIds = new ArrayList<>();
+
+        Subscription subscription = getDefaultClient().subscribeToAll(new SubscriptionListener() {
+            @Override
+            public void onEvent(Subscription subscription, ResolvedEvent event) {
+                synchronized (receivedStreamIds) {
+                    receivedStreamIds.add(event.getOriginalEvent().getStreamId());
+                }
+                count.incrementAndGet();
+                receivedEvents.countDown();
+            }
+
+            @Override
+            public void onCancelled(Subscription subscription, Throwable throwable) {
+                if (throwable == null)
+                    return;
+
+                Assertions.fail(throwable);
+            }
+        }, SubscribeToAllOptions.get()
+                .filter(SubscriptionFilter
+                        .newBuilder()
+                        .addStreamNamePrefix(streamPrefix)
+                        .build())).get();
+
+        getDefaultClient().appendToStream(matchingStream, generateEvents(expectedCount, eventType).iterator()).get();
+        getDefaultClient().appendToStream(nonMatchingStream, generateEvents(3, eventType).iterator()).get();
+
+        receivedEvents.await();
+
+        Thread.sleep(2000);
+
+        Assertions.assertEquals(expectedCount, count.get(),
+                "Filter should only match streams starting with the prefix, not streams containing it");
+
+        synchronized (receivedStreamIds) {
+            for (String streamId : receivedStreamIds) {
+                Assertions.assertTrue(streamId.startsWith(streamPrefix),
+                        "Received event from stream '" + streamId + "' which does not start with prefix '" + streamPrefix + "'");
+            }
+        }
+
+        subscription.stop();
+    }
+
     @Test
     default void testCancellingSubscriptionShouldNotRaiseAnException() throws Throwable {
         KurrentDBClient client = getDefaultClient();