feat: make the tracking publisher honor user:pass@ or token@ credentials in NATS_URL by lifting them into async-nats ConnectOptions, since async-nats ignores URL userinfo unlike the Go client, and redact the userinfo from the connect log line

docs/content/docs/development/configuration.mdx

@@ -258,7 +258,7 @@ On `filesystem`, a remote worker writes blobs to its own disk rather than a volu
 | Variable | What it does | Default | Restart needed |
 |---|---|---|---|
 | `EVENTBUS_PROVIDER` | `nats` or `kafka`. Kafka needs images built with `GO_TAGS=kafka` | `nats` under compose, `kafka` for a bare binary | yes |
-| `NATS_URL` | JetStream address | `nats://nats:4222` | yes |
+| `NATS_URL` | JetStream address. Credentials in the URL are honored by every service, including the Rust tracking publisher: `nats://user:pass@host:4222` for a user, `nats://token@host:4222` for a token, `tls://` for TLS | `nats://nats:4222` | yes |
 | `NATS_STREAM_NAME`, `NATS_SUBJECT_PREFIX` | Stream and subject naming | `warmbly` | yes |
 | `KAFKA_BOOTSTRAP_SERVERS` | Broker list when `EVENTBUS_PROVIDER=kafka` | unset | yes |
 | `KAFKA_SASL_USERNAME`, `KAFKA_SASL_PASSWORD` | Broker credentials | unset | yes |

tracking/src/nats.rs

@@ -17,12 +17,22 @@ pub struct NatsProducer {
 
 impl NatsProducer {
     pub async fn new(config: &Config) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
-        let client = async_nats::connect(&config.nats_url).await?;
+        // async-nats ignores credentials in the URL, so lift `user:pass@` or
+        // `token@` into the options the way the Go client does on its own.
+        let addr: async_nats::ServerAddr = config.nats_url.parse()?;
+        let mut opts = async_nats::ConnectOptions::new();
+        if let Some(user) = addr.username() {
+            opts = match addr.password() {
+                Some(pass) => opts.user_and_password(user.to_string(), pass.to_string()),
+                None => opts.token(user.to_string()),
+            };
+        }
+        let client = opts.connect(addr).await?;
         let js = jetstream::new(client);
         let subject = format!("{}.{}", config.nats_subject_prefix, config.kafka_topic);
         tracing::info!(
             "NATS producer connected to {}, publishing to subject {}",
-            config.nats_url,
+            redact_url(&config.nats_url),
             subject
         );
         Ok(Self { js, subject })
@@ -64,3 +74,28 @@ impl NatsProducer {
         }
     }
 }
+
+// redact_url drops any userinfo so credentials never reach the logs.
+fn redact_url(raw: &str) -> String {
+    match (raw.find("://"), raw.rfind('@')) {
+        (Some(scheme_end), Some(at)) if at > scheme_end => {
+            format!("{}://***@{}", &raw[..scheme_end], &raw[at + 1..])
+        }
+        _ => raw.to_string(),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::redact_url;
+
+    #[test]
+    fn redacts_userinfo_only() {
+        assert_eq!(redact_url("nats://127.0.0.1:4222"), "nats://127.0.0.1:4222");
+        assert_eq!(
+            redact_url("tls://tok3n@nats.example.com:4222"),
+            "tls://***@nats.example.com:4222"
+        );
+        assert_eq!(redact_url("nats://u:p@h:4222"), "nats://***@h:4222");
+    }
+}