feat: QueueType (QUEUE/STREAM) mode for @RqueueListener + NATS stream provisioning

- Add QueueType enum (QUEUE = WorkQueue retention, STREAM = Limits/fan-out)
- Add queueMode() attribute to @RqueueListener, default QUEUE
- Thread QueueType through MappingInformation → QueueDetail builder
- MessageBroker.onQueueRegistered() hook; JetStreamMessageBroker overrides it
  to provision the stream immediately on registerQueue() calls (producer-only mode)
- NatsProvisioner.ensureStream(name, subjects, QueueType) picks WorkQueue vs
  Limits retention; warns and preserves existing config on mismatch
- NatsStreamValidator and JetStreamMessageBroker pass q.getType() to provisioner
- RqueueRedisConfigImportSelector: lazy-load RqueueRedisListenerConfig via
  ImportSelector so excluding rqueue-redis no longer crashes Spring at startup
- Fix JetStreamMessageBrokerPeekIT and IndependentConsumersIT to use QueueType
  instead of the now-bypassed setRetention() config workaround
- Add JetStreamQueueModeIT: 5 E2E contracts covering stream retention, consumer
  reuse/position preservation, competing-consumer delivery, and fan-out delivery

Assisted-By: Claude Code

Author: sonus21

nats-task-v2.md

@@ -0,0 +1,72 @@
+# NATS backend — v2 task tracker
+
+## v1 status: COMPLETE
+
+All v1 items are done and 360 unit tests pass. Branch `nats-backend` is ready to merge.
+
+---
+
+## v2 pending items
+
+### 1. Web dashboard — NATS gaps
+
+Controllers are no longer Redis-gated but several operations throw `BackendCapabilityException` (HTTP 501) on NATS. The front-end should hide unsupported panels proactively instead of relying on 501s.
+
+- Expose `GET /rqueue/api/capabilities` returning the `Capabilities` record so the UI can conditionally hide panels.
+- Extend `Capabilities` with dashboard-op flags: `supportsCharts`, `supportsMessageBrowse`, `supportsAdminMove`.
+- Wire the flags into Pebble templates (scheduled panel, cron jobs panel, chart panel already have `hideScheduledPanel` / `hideCronJobs` hooks in `DataViewResponse`).
+
+Affected services that throw on NATS today:
+- `RqueueDashboardChartServiceImpl` — time-series charts (no equivalent in JetStream)
+- `RqueueUtilityServiceImpl` — move/enqueue admin ops
+- `NatsMessageBrowsingRepository.viewData` — positional message browse
+
+### 2. Reactive listener container
+
+Only the enqueue side is reactive in v1. The listener/pop side still uses blocking `BrokerMessagePoller` threads.
+
+- Implement a `ReactiveMessagePoller` using `js.publishAsync` + Project Reactor for the NATS path.
+- Gate behind `@Conditional(ReactiveEnabled.class)` + `NatsBackendCondition`.
+
+### 3. Delayed / scheduled / cron messages on NATS
+
+`enqueueWithDelay` throws `UnsupportedOperationException` in v1. Options:
+
+- Use NATS JetStream `MaxAge` + a separate "delay bucket" stream per delay tier (coarse buckets: 1s, 5s, 30s, 5m, 1h).
+- Or implement a lightweight delay-scheduler sidecar using KV TTL expiry events.
+
+### 4. `priorityGroup` weighting on NATS
+
+In v1, cross-queue `priorityGroup` weighting logs a boot WARN and is not honored. `BrokerMessagePoller` spawns one thread per `(queue, consumerName, priority)` triple at fixed weight.
+
+- Implement weighted round-robin across pollers sharing the same `priorityGroup`.
+
+### 5. Elastic concurrency (`@RqueueListener.concurrency min < max`)
+
+Falls back to fixed `max` on NATS in v1. Implement auto-scaling poller count based on queue depth via `MessageBroker.size()`.
+
+### 6. `@RqueueHandler(primary)` on NATS
+
+Ignored in v1 with a single boot WARN. On NATS all handler methods are dispatched independently (one consumer per method). `primary` could select the default handler when message type is ambiguous.
+
+### 7. Spring Boot configuration metadata (source annotation processor)
+
+`rqueue.nats.auto-create-kv-buckets` and sibling properties appear in the built metadata JSON but only from the compiled artifact. Add `spring-boot-configuration-processor` to the starter's `annotationProcessor` deps so IDEs pick up descriptions and defaults without a pre-build step.
+
+### 8. `RqueueStringDao` javadoc
+
+Mark as Redis-internal in the interface javadoc so future contributors don't try to add a NATS impl.
+
+---
+
+## Local verification commands
+
+```bash
+./gradlew :rqueue-core:test :rqueue-redis:test :rqueue-web:test :rqueue-nats:test -DincludeTags=unit
+./gradlew :rqueue-spring-boot-starter:test --tests "com.github.sonus21.rqueue.spring.boot.integration.NatsBackendEndToEndIT"
+./gradlew :rqueue-nats:test -DincludeTags=nats
+```
+
+## Commit-rule reminder
+
+`CLAUDE.md` forbids `Co-Authored-By:` for any AI tool. Use `Assisted-By: Claude Code` only.

rqueue-core/src/main/java/com/github/sonus21/rqueue/annotation/RqueueListener.java

@@ -16,6 +16,7 @@
 
 package com.github.sonus21.rqueue.annotation;
 
+import com.github.sonus21.rqueue.enums.QueueType;
 import com.github.sonus21.rqueue.utils.Constants;
 import java.lang.annotation.Documented;
 import java.lang.annotation.ElementType;
@@ -204,4 +205,23 @@
    * @return durable consumer name override; empty for "auto-generate"
    */
   String consumerName() default "";
+
+  /**
+   * Delivery mode for this queue.
+   *
+   * <ul>
+   *   <li>{@link QueueType#QUEUE} (default) — competing-consumer semantics: each
+   *       message is delivered to exactly one listener instance. Maps to a JetStream
+   *       {@code WorkQueue} stream. Multiple listeners share the load.
+   *   <li>{@link QueueType#STREAM} — fan-out semantics: every independent listener
+   *       group receives a copy of each message. Maps to a JetStream {@code Limits} stream.
+   *       Each listener must use a distinct {@link #consumerName()} so JetStream tracks its
+   *       own position independently.
+   * </ul>
+   *
+   * <p>The Redis backend ignores this attribute (Redis queue names already encode the routing).
+   *
+   * @return queue delivery mode
+   */
+  QueueType queueMode() default QueueType.QUEUE;
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/impl/BaseMessageSender.java

@@ -29,6 +29,7 @@
 import com.github.sonus21.rqueue.core.RqueueMessageIdGenerator;
 import com.github.sonus21.rqueue.core.RqueueMessageTemplate;
 import com.github.sonus21.rqueue.core.impl.MessageSweeper.MessageDeleteRequest;
+import com.github.sonus21.rqueue.enums.QueueType;
 import com.github.sonus21.rqueue.exception.DuplicateMessageException;
 import com.github.sonus21.rqueue.listener.QueueDetail;
 import com.github.sonus21.rqueue.models.db.MessageMetadata;
@@ -99,12 +100,13 @@ protected Object enqueue(
   }
 
   /**
-   * Priority-aware enqueue. When a non-Redis {@link com.github.sonus21.rqueue.core.spi.MessageBroker}
-   * is set on the underlying {@link RqueueMessageTemplate} (i.e. capabilities advertise
-   * {@code !usesPrimaryHandlerDispatch}) this routes the publish through
+   * Priority-aware enqueue. When a non-Redis
+   * {@link com.github.sonus21.rqueue.core.spi.MessageBroker} is set on the underlying
+   * {@link RqueueMessageTemplate} (i.e. capabilities advertise {@code !usesPrimaryHandlerDispatch})
+   * this routes the publish through
    * {@link com.github.sonus21.rqueue.core.spi.MessageBroker#enqueue(QueueDetail, String,
-   * RqueueMessage)} so backends like NATS can publish to a priority-specific subject. Otherwise
-   * the existing Redis-shaped path is used; Redis already encodes priority in the queue name so
+   * RqueueMessage)} so backends like NATS can publish to a priority-specific subject. Otherwise the
+   * existing Redis-shaped path is used; Redis already encodes priority in the queue name so
    * {@code priority} is ignored.
    */
   protected Object enqueue(
@@ -218,7 +220,8 @@ protected Object deleteAllMessages(QueueDetail queueDetail) {
             MessageDeleteRequest.builder().queueDetail(queueDetail).build());
   }
 
-  protected void registerQueueInternal(String queueName, String... priorities) {
+  protected void registerQueueInternal(String queueName, QueueType type,
+      String... priorities) {
     validateQueue(queueName);
     notNull(priorities, "priorities cannot be null");
     Map<String, Integer> priorityMap = new HashMap<>();
@@ -236,8 +239,10 @@ protected void registerQueueInternal(String queueName, String... priorities) {
         .processingQueueName(rqueueConfig.getProcessingQueueName(queueName))
         .processingQueueChannelName(rqueueConfig.getProcessingQueueChannelName(queueName))
         .priority(priorityMap)
+        .type(type)
         .build();
     EndpointRegistry.register(queueDetail);
+    notifyBrokerQueueRegistered(queueDetail);
     for (String priority : priorities) {
       String suffix = PriorityUtils.getSuffix(priority);
       queueDetail = QueueDetail.builder()
@@ -252,6 +257,14 @@ protected void registerQueueInternal(String queueName, String... priorities) {
           .priority(Collections.singletonMap(DEFAULT_PRIORITY_KEY, 1))
           .build();
       EndpointRegistry.register(queueDetail);
+      notifyBrokerQueueRegistered(queueDetail);
+    }
+  }
+
+  private void notifyBrokerQueueRegistered(QueueDetail queueDetail) {
+    com.github.sonus21.rqueue.core.spi.MessageBroker broker = messageTemplate.getMessageBroker();
+    if (broker != null) {
+      broker.onQueueRegistered(queueDetail);
     }
   }
 }

rqueue-core/src/main/java/com/github/sonus21/rqueue/core/spi/MessageBroker.java

@@ -46,6 +46,13 @@ default void enqueue(QueueDetail q, String priority, RqueueMessage m) {
 
   void enqueueWithDelay(QueueDetail q, RqueueMessage m, long delayMs);
 
+  /**
+   * Called by {@code RqueueEndpointManager.registerQueue} after a queue is added to the registry.
+   * Backends that need to provision resources (e.g. JetStream streams) at registration time should
+   * override this. The default is a no-op so Redis and other backends are unaffected.
+   */
+  default void onQueueRegistered(QueueDetail q) {}
+
   /**
    * Reactive variant of {@link #enqueue(QueueDetail, RqueueMessage)}. The default falls back to the
    * blocking implementation wrapped in {@code Mono.fromRunnable}; backends with native async

rqueue-core/src/main/java/com/github/sonus21/rqueue/enums/QueueType.java

@@ -0,0 +1,12 @@
+package com.github.sonus21.rqueue.enums;
+
+public enum QueueType {
+  /**
+   * WorkQueue semantics: competing consumers, each message delivered to exactly one listener.
+   */
+  QUEUE,
+  /**
+   * Stream/fan-out semantics: every independent listener group receives all messages.
+   */
+  STREAM
+}

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/MappingInformation.java

@@ -19,6 +19,7 @@
 import static com.github.sonus21.rqueue.utils.Constants.DELTA_BETWEEN_RE_ENQUEUE_TIME;
 import static com.github.sonus21.rqueue.utils.Constants.MIN_EXECUTION_TIME;
 
+import com.github.sonus21.rqueue.enums.QueueType;
 import com.github.sonus21.rqueue.models.Concurrency;
 import java.util.Map;
 import java.util.Set;
@@ -50,6 +51,9 @@ class MappingInformation implements Comparable<MappingInformation> {
   private final int batchSize;
   private final Set<Class<? extends Throwable>> doNotRetry;
 
+  @Builder.Default
+  private final QueueType queueType = QueueType.QUEUE;
+
   @Override
   public String toString() {
     return String.join(", ", queueNames);

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/QueueDetail.java

@@ -17,6 +17,7 @@
 package com.github.sonus21.rqueue.listener;
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.github.sonus21.rqueue.enums.QueueType;
 import com.github.sonus21.rqueue.models.Concurrency;
 import com.github.sonus21.rqueue.models.SerializableBase;
 import com.github.sonus21.rqueue.models.db.DeadLetterQueue;
@@ -260,11 +261,6 @@ public int resolvedMaxDeliver(int fallback) {
     return natsMaxDeliverOverride != null ? natsMaxDeliverOverride : fallback;
   }
 
-  public enum QueueType {
-    QUEUE,
-    STREAM
-  }
-
   public boolean isDoNotRetryError(Throwable throwable) {
     if (Objects.isNull(throwable)) {
       return false;

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/RqueueMessageHandler.java

@@ -334,6 +334,7 @@ private MappingInformation getMappingInformation(RqueueListener rqueueListener)
         .batchSize(batchSize)
         .doNotRetry(new HashSet<>(Arrays.asList(rqueueListener.doNotRetry())))
         .consumerName(natsConsumerName.isEmpty() ? null : natsConsumerName)
+        .queueType(rqueueListener.queueMode())
         .build();
     if (mappingInformation.isValid()) {
       return mappingInformation;

rqueue-core/src/main/java/com/github/sonus21/rqueue/listener/RqueueMessageListenerContainer.java

@@ -503,6 +503,7 @@ private List<QueueDetail> getQueueDetail(String queue, MappingInformation mappin
         .priorityGroup(priorityGroup)
         .doNotRetry(mappingInformation.getDoNotRetry())
         .consumerName(mappingInformation.getConsumerName())
+        .type(mappingInformation.getQueueType())
         .build();
     List<QueueDetail> queueDetails;
     if (queueDetail.getPriority().size() <= 1) {

rqueue-nats/src/main/java/com/github/sonus21/rqueue/nats/internal/NatsProvisioner.java

@@ -9,6 +9,7 @@
  */
 package com.github.sonus21.rqueue.nats.internal;
 
+import com.github.sonus21.rqueue.enums.QueueType;
 import com.github.sonus21.rqueue.nats.RqueueNatsConfig;
 import com.github.sonus21.rqueue.nats.RqueueNatsException;
 import io.nats.client.Connection;
@@ -23,6 +24,7 @@
 import io.nats.client.api.DeliverPolicy;
 import io.nats.client.api.KeyValueConfiguration;
 import io.nats.client.api.KeyValueStatus;
+import io.nats.client.api.RetentionPolicy;
 import io.nats.client.api.StreamConfiguration;
 import io.nats.client.api.StreamInfo;
 import java.io.IOException;
@@ -118,11 +120,29 @@ public KeyValue ensureKv(String bucketName, Duration ttl)
   // ---- Stream provisioning ----------------------------------------------
 
   /**
-   * Ensure a JetStream stream exists with the given subjects. Hits the NATS backend at most once
-   * per stream name per process lifetime; subsequent calls return immediately from the in-process
-   * cache.
+   * Ensure a JetStream stream exists with the given subjects, using {@link QueueType#QUEUE}
+   * (WorkQueue retention) as the default. Callers that have a {@link QueueType} available should
+   * use {@link #ensureStream(String, List, QueueType)} instead.
    */
   public void ensureStream(String streamName, List<String> subjects) {
+    ensureStream(streamName, subjects, QueueType.QUEUE);
+  }
+
+  /**
+   * Ensure a JetStream stream exists with the given subjects and retention policy derived from
+   * {@code queueType}:
+   * <ul>
+   *   <li>{@link QueueType#QUEUE} — {@link io.nats.client.api.RetentionPolicy#WorkQueue}: each
+   *       message is delivered to exactly one consumer; competing-consumer semantics.
+   *   <li>{@link QueueType#STREAM} — {@link io.nats.client.api.RetentionPolicy#Limits}: every
+   *       independent durable consumer group receives all messages; stream/fan-out semantics.
+   * </ul>
+   *
+   * <p>Hits the NATS backend at most once per stream name per process lifetime; subsequent calls
+   * return immediately from the in-process cache. If the stream already exists with a different
+   * retention policy, a WARNING is logged and the existing config is left untouched.
+   */
+  public void ensureStream(String streamName, List<String> subjects, QueueType queueType) {
     if (streamsDone.contains(streamName)) {
       return;
     }
@@ -133,6 +153,9 @@ public void ensureStream(String streamName, List<String> subjects) {
       }
       try {
         StreamInfo existing = safeGetStreamInfo(streamName);
+        RetentionPolicy desired = queueType == QueueType.STREAM
+            ? RetentionPolicy.Limits
+            : RetentionPolicy.WorkQueue;
         if (existing == null) {
           if (!config.isAutoCreateStreams()) {
             throw new RqueueNatsException(
@@ -144,7 +167,7 @@ public void ensureStream(String streamName, List<String> subjects) {
               .subjects(subjects)
               .replicas(sd.getReplicas())
               .storageType(sd.getStorage())
-              .retentionPolicy(sd.getRetention())
+              .retentionPolicy(desired)
               .duplicateWindow(sd.getDuplicateWindow())
               .compressionOption(CompressionOption.S2);
           if (sd.getMaxMsgs() > 0) {
@@ -159,6 +182,15 @@ public void ensureStream(String streamName, List<String> subjects) {
             b.maxAge(sd.getMaxAge());
           }
           jsm.addStream(b.build());
+        } else {
+          RetentionPolicy actual = existing.getConfiguration().getRetentionPolicy();
+          if (actual != desired) {
+            log.log(
+                Level.WARNING,
+                "Stream ''{0}'' exists with retention={1} but queueMode requires retention={2}"
+                    + " — leaving existing config in place.",
+                new Object[] {streamName, actual, desired});
+          }
         }
       } catch (IOException | JetStreamApiException e) {
         throw new RqueueNatsException(