autoconnect_settings/
app_state.rs

1use std::{sync::Arc, time::Duration};
2
3#[cfg(feature = "bigtable")]
4use autopush_common::db::bigtable::BigTableClientImpl;
5#[cfg(feature = "postgres")]
6use autopush_common::db::postgres::PgClientImpl;
7#[cfg(feature = "redis")]
8use autopush_common::db::redis::RedisClientImpl;
9use cadence::StatsdClient;
10use config::ConfigError;
11use fernet::{Fernet, MultiFernet};
12use tokio::sync::RwLock;
13
14use autoconnect_common::{
15    broadcast::BroadcastChangeTracker, megaphone::init_and_spawn_megaphone_updater,
16    registry::ClientRegistry,
17};
18use autopush_common::db::{DbSettings, StorageType, client::DbClient};
19#[cfg(feature = "reliable_report")]
20use autopush_common::reliability::PushReliability;
21
22use crate::{ENV_PREFIX, Settings};
23
24#[derive(Clone)]
25pub struct AppState {
26    /// Handle to the data storage object
27    pub db: Box<dyn DbClient>,
28    pub metrics: Arc<StatsdClient>,
29    pub http: reqwest::Client,
30
31    /// Encryption object for the endpoint URL
32    pub fernet: MultiFernet,
33    /// The connected WebSocket clients
34    pub clients: Arc<ClientRegistry>,
35    /// The Megaphone Broadcast change tracker
36    pub broadcaster: Arc<RwLock<BroadcastChangeTracker>>,
37
38    pub settings: Settings,
39    pub router_url: String,
40    pub endpoint_url: String,
41
42    #[cfg(feature = "reliable_report")]
43    pub reliability: Arc<PushReliability>,
44}
45
46impl AppState {
47    pub fn from_settings(settings: Settings) -> Result<Self, ConfigError> {
48        let crypto_key = &settings.crypto_key;
49        if !(crypto_key.starts_with('[') && crypto_key.ends_with(']')) {
50            return Err(ConfigError::Message(format!(
51                "Invalid {ENV_PREFIX}_CRYPTO_KEY"
52            )));
53        }
54        let crypto_key = &crypto_key[1..crypto_key.len() - 1];
55        debug!("🔐 Fernet keys: {:?}", &crypto_key);
56        let fernets: Vec<Fernet> = crypto_key
57            .split(',')
58            .map(|s| s.trim().to_string())
59            .map(|key| {
60                Fernet::new(&key).unwrap_or_else(|| panic!("Invalid {ENV_PREFIX}_CRYPTO_KEY"))
61            })
62            .collect();
63        let fernet = MultiFernet::new(fernets);
64        let metrics = autopush_common::metrics::builder(
65            &settings.statsd_label,
66            &settings.statsd_host,
67            settings.statsd_port,
68        )
69        .map_err(|e| ConfigError::Message(e.to_string()))?
70        // Temporary tag to distinguish from the legacy autopush(connect)
71        .with_tag("autoconnect", "true")
72        .build();
73        let metrics = Arc::new(metrics);
74
75        let db_settings = DbSettings {
76            dsn: settings.db_dsn.clone(),
77            db_settings: settings.db_settings.clone(),
78        };
79        let storage_type = StorageType::from_dsn(&db_settings.dsn);
80
81        #[allow(unused)]
82        let db: Box<dyn DbClient> = match storage_type {
83            #[cfg(feature = "bigtable")]
84            StorageType::BigTable => {
85                let client = BigTableClientImpl::new(metrics.clone(), &db_settings)
86                    .map_err(|e| ConfigError::Message(e.to_string()))?;
87                Box::new(client)
88            }
89            #[cfg(feature = "postgres")]
90            StorageType::Postgres => {
91                let client = PgClientImpl::new(metrics.clone(), &db_settings)
92                    .map_err(|e| ConfigError::Message(e.to_string()))?;
93                Box::new(client)
94            }
95            #[cfg(feature = "redis")]
96            StorageType::Redis => Box::new(
97                RedisClientImpl::new(metrics.clone(), &db_settings)
98                    .map_err(|e| ConfigError::Message(e.to_string()))?,
99            ),
100            _ => panic!(
101                "Invalid Storage type {:?}. Check {}__DB_DSN.",
102                storage_type,
103                ENV_PREFIX.to_uppercase()
104            ),
105        };
106
107        #[cfg(feature = "reliable_report")]
108        let reliability = Arc::new(
109            PushReliability::new(
110                &settings.reliability_dsn,
111                db.clone(),
112                &metrics,
113                settings.reliability_retry_count,
114            )
115            .map_err(|e| {
116                ConfigError::Message(format!("Could not start Reliability connection: {e:?}"))
117            })?,
118        );
119        let http = reqwest::Client::builder()
120            .timeout(Duration::from_secs(1))
121            .pool_max_idle_per_host(settings.pool_max_idle_per_host)
122            .pool_idle_timeout(Duration::from_secs(settings.pool_idle_timeout_secs))
123            .build()
124            .unwrap_or_else(|e| panic!("Error while building reqwest::Client: {e}"));
125        let broadcaster = Arc::new(RwLock::new(BroadcastChangeTracker::new(Vec::new())));
126
127        let router_url = settings.router_url();
128        let endpoint_url = settings.endpoint_url();
129
130        Ok(Self {
131            db,
132            metrics,
133            http,
134            fernet,
135            clients: Arc::new(ClientRegistry::with_channel_capacity(
136                settings.client_channel_capacity,
137            )),
138            broadcaster,
139            settings,
140            router_url,
141            endpoint_url,
142            #[cfg(feature = "reliable_report")]
143            reliability,
144        })
145    }
146
147    /// Initialize the `BroadcastChangeTracker`
148    ///
149    /// Via `autoconnect_common::megaphone::init_and_spawn_megaphone_updater`
150    pub async fn init_and_spawn_megaphone_updater(&self) -> Result<(), ConfigError> {
151        let Some(ref url) = self.settings.megaphone_api_url else {
152            return Ok(());
153        };
154        init_and_spawn_megaphone_updater(
155            &self.broadcaster,
156            &self.http,
157            &self.metrics,
158            url,
159            self.settings.megaphone_poll_interval,
160        )
161        .await
162        .map_err(|e| ConfigError::Message(e.to_string()))?;
163        Ok(())
164    }
165}
166
167/// For tests
168#[cfg(debug_assertions)]
169impl Default for AppState {
170    fn default() -> Self {
171        Self::from_settings(Settings::test_settings()).unwrap()
172    }
173}
autoconnect_settings/app_state.rs

autoconnect_settings/
app_state.rs
