fix(svc): authenticate to nats with the configured token (#2581)

deploy/self-hosting/docker-compose.yml

@@ -26,6 +26,7 @@ x-fluxer-env: &fluxer-env
   FLUXER_KV_URL: redis://valkey:6379/0
   FLUXER_NATS_URL: nats://nats:4222
   FLUXER_NATS_JETSTREAM_URL: nats://nats:4222
+  FLUXER_NATS_AUTH_TOKEN: ${FLUXER_NATS_AUTH_TOKEN:-}
   FLUXER_SVC_NATS_URL: nats://nats:4222
   FLUXER_SVC_SHARD_COUNT: "1"
 

fluxer_docs/src/content/docs/operator/configuration.mdx

@@ -581,7 +581,7 @@ Default `nats://127.0.0.1:4222`. The JetStream address. Read by `api` and `worke
 
 #### `FLUXER_NATS_AUTH_TOKEN`
 
-Default empty. NATS authentication. Read by `api`, `worker`, and `gateway`. The shipped NATS runs without authentication.
+Default empty. NATS authentication. Read by `api`, `worker`, `gateway`, and the five internal services. The shipped NATS runs without authentication, and Compose forwards this name to every container that connects to it.
 
 #### `FLUXER_SVC_NATS_URL`
 

fluxer_gifs/src/main.rs

@@ -15,7 +15,8 @@ use shard_impl::GifsShard;
 async fn main() -> anyhow::Result<()> {
     fluxer_svc::init_tracing();
     let config = ServiceConfig::from_env()?;
-    let transport = NatsTransport::connect(&config.nats_url).await?;
+    let transport =
+        NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?;
 
     tracing::info!(
         service = config.service_name,

fluxer_messages/src/main.rs

@@ -9,7 +9,8 @@ use fluxer_svc::transport::NatsTransport;
 async fn main() -> anyhow::Result<()> {
     fluxer_svc::init_tracing();
     let config = ServiceConfig::from_env()?;
-    let transport = NatsTransport::connect(&config.nats_url).await?;
+    let transport =
+        NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?;
 
     tracing::info!(
         service = config.service_name,

fluxer_snowflakes/src/main.rs

@@ -14,7 +14,8 @@ use types::SERVICE_NAME;
 async fn main() -> anyhow::Result<()> {
     fluxer_svc::init_tracing();
     let config = ServiceConfig::from_env()?;
-    let transport = NatsTransport::connect(&config.nats_url).await?;
+    let transport =
+        NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?;
     tracing::info!(
         service = SERVICE_NAME,
         mode = ?config.mode,

fluxer_svc/src/config.rs

@@ -17,6 +17,7 @@ pub struct ServiceConfig {
     pub shard_count: u32,
     pub listen_addr: SocketAddr,
     pub nats_url: String,
+    pub nats_auth_token: Option<String>,
     pub cache_max_entries: u64,
     pub cache_ttl: Duration,
     pub cache_hard_ttl: Duration,
@@ -105,6 +106,8 @@ impl ServiceConfig {
         let nats_url = optional_from(&get, "FLUXER_SVC_NATS_URL")
             .unwrap_or_else(|| "nats://127.0.0.1:4222".to_owned());
 
+        let nats_auth_token = optional_from(&get, "FLUXER_NATS_AUTH_TOKEN");
+
         let cache_ttl_ms = optional_from(&get, "FLUXER_SVC_CACHE_TTL_MS")
             .map(|v| v.parse::<u64>())
             .transpose()?
@@ -165,6 +168,7 @@ impl ServiceConfig {
             shard_count,
             listen_addr: format!("{listen_host}:{listen_port}").parse()?,
             nats_url,
+            nats_auth_token,
             cache_max_entries: optional_from(&get, "FLUXER_SVC_CACHE_MAX_ENTRIES")
                 .map(|v| v.parse::<u64>())
                 .transpose()?

fluxer_svc/src/lib.rs

@@ -33,7 +33,9 @@ where
 {
     init_tracing();
     let config = config::ServiceConfig::from_env()?;
-    let transport = transport::NatsTransport::connect(&config.nats_url).await?;
+    let transport =
+        transport::NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref())
+            .await?;
     tracing::info!(
         service = config.service_name,
         mode = ?config.mode,

fluxer_svc/src/router.rs

@@ -868,6 +868,7 @@ mod tests {
             shard_count: 1,
             listen_addr: "127.0.0.1:0".parse().unwrap(),
             nats_url: "memory".to_owned(),
+            nats_auth_token: None,
             cache_max_entries: 100,
             cache_ttl: Duration::from_secs(30),
             cache_hard_ttl: Duration::from_secs(600),

fluxer_svc/src/shard.rs

@@ -417,6 +417,7 @@ mod tests {
             shard_count: 1,
             listen_addr: "127.0.0.1:0".parse().unwrap(),
             nats_url: "memory".to_owned(),
+            nats_auth_token: None,
             cache_max_entries: 100,
             cache_ttl: Duration::from_secs(30),
             cache_hard_ttl: Duration::from_secs(600),

fluxer_svc/src/transport.rs

@@ -77,7 +77,7 @@ pub struct NatsMessage {
 }
 
 impl NatsTransport {
-    pub async fn connect(url: &str) -> anyhow::Result<Self> {
+    pub async fn connect(url: &str, auth_token: Option<&str>) -> anyhow::Result<Self> {
         let reconnect_notify = Arc::new(Notify::new());
 
         let event_notify = reconnect_notify.clone();
@@ -112,6 +112,11 @@ impl NatsTransport {
             }
         });
 
+        let options = match auth_token {
+            Some(token) => options.token(token.to_owned()),
+            None => options,
+        };
+
         let client = options
             .subscription_capacity(NATS_SUBSCRIPTION_CAPACITY)
             .connect(url)

fluxer_unfurl/src/main.rs

@@ -27,7 +27,8 @@ use shard_impl::UnfurlShard;
 async fn main() -> anyhow::Result<()> {
     fluxer_svc::init_tracing();
     let config = ServiceConfig::from_env()?;
-    let transport = NatsTransport::connect(&config.nats_url).await?;
+    let transport =
+        NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?;
 
     tracing::info!(
         service = config.service_name,

fluxer_users/src/main.rs

@@ -13,7 +13,8 @@ use shard_impl::UsersShard;
 async fn main() -> anyhow::Result<()> {
     fluxer_svc::init_tracing();
     let config = ServiceConfig::from_env()?;
-    let transport = NatsTransport::connect(&config.nats_url).await?;
+    let transport =
+        NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?;
 
     tracing::info!(
         service = config.service_name,