fix: wire TopicResolver into KafkaTransport — auto-discover when topics empty

Author: catinspace-au

src/transport/kafka/mod.rs

@@ -62,13 +62,15 @@ pub use admin::{KafkaAdmin, TopicInfo};
 pub use config::{
     DEVTEST_PROFILE, HIGH_THROUGHPUT_CONSUMER_DEFAULTS, KafkaConfig, KafkaProfile,
     LOW_LATENCY_CONSUMER_DEFAULTS, PRODUCER_DEFAULTS, PRODUCER_DEVTEST, PRODUCER_EXACTLY_ONCE,
-    PRODUCER_HIGH_THROUGHPUT, PRODUCER_LOW_LATENCY, PRODUCTION_PROFILE, merge_with_overrides,
+    PRODUCER_HIGH_THROUGHPUT, PRODUCER_LOW_LATENCY, PRODUCTION_PROFILE, SuppressionRule,
+    merge_with_overrides,
 };
 pub use metrics::{
     BrokerMetrics, KafkaMetrics, StatsContext, healthy_broker_count, total_consumer_lag,
 };
 pub use producer::{KafkaProducer, ProducerMetrics, ProducerProfile};
 pub use token::KafkaToken;
+pub use topic_resolver::{TopicRefreshHandle, TopicResolver};
 
 use super::error::{TransportError, TransportResult};
 use super::traits::{TransportBase, TransportReceiver, TransportSender};
@@ -215,8 +217,23 @@ impl KafkaTransport {
             .create_with_context(StatsContext::new())
             .map_err(|e| TransportError::Connection(format!("Failed to create consumer: {e}")))?;
 
+        // Resolve effective topics: use explicit list or auto-discover from broker
+        let effective_topics = if config.topics.is_empty() {
+            tracing::info!("Topics empty — auto-discovering from broker");
+            let resolver = topic_resolver::TopicResolver::new(config)?;
+            let discovered = resolver.resolve()?;
+            if discovered.is_empty() {
+                return Err(TransportError::Config(
+                    "Auto-discovery found no matching topics".into(),
+                ));
+            }
+            discovered
+        } else {
+            config.topics.clone()
+        };
+
         // Subscribe to topics
-        let subscribed_topics = config.topics.clone();
+        let subscribed_topics = effective_topics;
         if !subscribed_topics.is_empty() {
             let topics: Vec<&str> = subscribed_topics.iter().map(String::as_str).collect();
             consumer

src/transport/kafka/topic_resolver.rs

@@ -146,6 +146,7 @@ impl std::fmt::Debug for TopicResolver {
 /// With the default rule (`_load` suppresses `_land`):
 /// - `auth_load` + `auth_land` → keeps `auth_load`, drops `auth_land`
 /// - `events_land` (no `events_load`) → kept
+#[must_use]
 pub fn apply_suppression_rules(topics: Vec<String>, rules: &[SuppressionRule]) -> Vec<String> {
     if rules.is_empty() {
         return topics;
@@ -161,16 +162,13 @@ pub fn apply_suppression_rules(topics: Vec<String>, rules: &[SuppressionRule]) -
             })
             .collect();
 
-        result = result
-            .into_iter()
-            .filter(|t| {
-                if let Some(base) = t.strip_suffix(rule.suppressed_suffix.as_str()) {
-                    !preferred_bases.contains(base)
-                } else {
-                    true
-                }
-            })
-            .collect();
+        result.retain(|t| {
+            if let Some(base) = t.strip_suffix(rule.suppressed_suffix.as_str()) {
+                !preferred_bases.contains(base)
+            } else {
+                true
+            }
+        });
     }
     result
 }
@@ -184,6 +182,7 @@ pub fn apply_suppression_rules(topics: Vec<String>, rules: &[SuppressionRule]) -
 /// - **Include**: if patterns exist, the topic MUST match at least one (OR).
 ///   Empty include list means all topics are accepted.
 /// - **Exclude**: the topic MUST NOT match any pattern (OR). Exclude wins over include.
+#[must_use]
 pub fn passes_filters(topic: &str, include: &[Regex], exclude: &[Regex]) -> bool {
     if !include.is_empty() && !include.iter().any(|r| r.is_match(topic)) {
         return false;
@@ -229,12 +228,11 @@ impl TopicRefreshHandle {
         if self.rx.has_changed().unwrap_or(false) {
             self.rx.mark_changed();
             let current = self.rx.borrow().clone();
-            if current != self.last_seen {
-                self.last_seen = current.clone();
-                Some(current)
-            } else {
-                None
+            if current == self.last_seen {
+                return None;
             }
+            self.last_seen.clone_from(&current);
+            Some(current)
         } else {
             None
         }
@@ -265,6 +263,7 @@ impl TopicResolver {
     /// so the initial topic list is taken from `resolve()` at construction time.
     ///
     /// If the shutdown token is cancelled, the background task exits cleanly.
+    #[must_use]
     pub fn start_refresh_loop(
         self,
         interval: std::time::Duration,
@@ -280,11 +279,11 @@ impl TopicResolver {
             loop {
                 tokio::select! {
                     biased;
-                    _ = shutdown.cancelled() => {
+                    () = shutdown.cancelled() => {
                         tracing::debug!("Topic refresh loop shutting down");
                         break;
                     }
-                    _ = ticker.tick() => {
+                    _tick = ticker.tick() => {
                         match self.resolve() {
                             Ok(new_topics) => {
                                 if tx.send(new_topics).is_err() {