unifly_api/controller/
refresh.rs

1use std::collections::HashSet;
2use std::sync::Arc;
3use std::time::Duration;
4
5use tokio_util::sync::CancellationToken;
6use tracing::{debug, warn};
7
8use crate::core_error::CoreError;
9use crate::model::Event;
10use crate::store::{DataStore, RefreshSnapshot, event_storage_key};
11
12use super::Controller;
13
14mod integration;
15mod merge;
16mod session;
17
18impl Controller {
19    /// Fetch all data from the controller and update the DataStore.
20    ///
21    /// Pulls devices, clients, and events from the controller APIs, converts
22    /// them to domain types, and applies them to the store. Events are
23    /// broadcast through the event channel after snapshot application.
24    pub async fn full_refresh(&self) -> Result<(), CoreError> {
25        let integration = self.inner.integration_client.lock().await.clone();
26        let site_id = *self.inner.site_id.lock().await;
27
28        if let (Some(integration), Some(site_id)) = (integration, site_id) {
29            self.refresh_integration_snapshot(integration, site_id)
30                .await?;
31        } else {
32            self.refresh_session_snapshot().await?;
33        }
34
35        debug!(
36            devices = self.inner.store.device_count(),
37            clients = self.inner.store.client_count(),
38            "data refresh complete"
39        );
40
41        Ok(())
42    }
43
44    async fn refresh_integration_snapshot(
45        &self,
46        integration: Arc<crate::IntegrationClient>,
47        site_id: uuid::Uuid,
48    ) -> Result<(), CoreError> {
49        let mut integration = integration::fetch(integration, site_id).await?;
50        let session_client = self.inner.session_client.lock().await.clone();
51        let session = session::fetch_optional(session_client).await;
52
53        merge::merge_session_clients(&mut integration.clients, &session.clients, &session.users);
54        merge::merge_session_devices(&mut integration.devices, &session.devices);
55
56        if !session.health.is_empty() {
57            let health = Arc::new(session.health);
58            self.inner
59                .store
60                .site_health
61                .send_modify(|current| *current = Arc::clone(&health));
62        }
63
64        let fresh_events = unseen_events(self.store(), &session.events);
65        self.inner
66            .store
67            .apply_integration_snapshot(RefreshSnapshot {
68                devices: integration.devices,
69                clients: integration.clients,
70                networks: integration.networks,
71                wifi: integration.wifi,
72                policies: integration.policies,
73                zones: integration.zones,
74                acls: integration.acls,
75                nat: session.nat,
76                dns: integration.dns,
77                vouchers: integration.vouchers,
78                sites: integration.sites,
79                events: session.events,
80                traffic_matching_lists: integration.traffic_matching_lists,
81                firewall_groups: session.firewall_groups,
82            });
83        self.publish_events(fresh_events);
84
85        Ok(())
86    }
87
88    async fn refresh_session_snapshot(&self) -> Result<(), CoreError> {
89        let session_client = self
90            .inner
91            .session_client
92            .lock()
93            .await
94            .clone()
95            .ok_or(CoreError::ControllerDisconnected)?;
96        let session = session::fetch_required(session_client).await?;
97        let fresh_events = unseen_events(self.store(), &session.events);
98
99        self.inner
100            .store
101            .apply_integration_snapshot(RefreshSnapshot {
102                devices: session.devices,
103                clients: session.clients,
104                networks: Vec::new(),
105                wifi: Vec::new(),
106                policies: Vec::new(),
107                zones: Vec::new(),
108                acls: Vec::new(),
109                nat: Vec::new(),
110                dns: Vec::new(),
111                vouchers: Vec::new(),
112                sites: session.sites,
113                events: session.events,
114                traffic_matching_lists: Vec::new(),
115                firewall_groups: session.firewall_groups,
116            });
117        self.publish_events(fresh_events);
118
119        Ok(())
120    }
121
122    fn publish_events(&self, events: Vec<Event>) {
123        for event in events {
124            let _ = self.inner.event_tx.send(Arc::new(event));
125        }
126    }
127}
128
129/// Periodically refresh data from the controller.
130pub(super) async fn refresh_task(
131    controller: Controller,
132    interval_secs: u64,
133    cancel: CancellationToken,
134) {
135    let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
136    interval.tick().await;
137
138    loop {
139        tokio::select! {
140            biased;
141            () = cancel.cancelled() => break,
142            _ = interval.tick() => {
143                if let Err(error) = controller.full_refresh().await {
144                    warn!(error = %error, "periodic refresh failed");
145                }
146            }
147        }
148    }
149}
150
151fn unseen_events(store: &DataStore, events: &[Event]) -> Vec<Event> {
152    let mut seen: HashSet<String> = store
153        .events_snapshot()
154        .iter()
155        .map(|event| event_storage_key(event))
156        .collect();
157
158    events
159        .iter()
160        .filter(|event| seen.insert(event_storage_key(event)))
161        .cloned()
162        .collect()
163}
unifly_api/controller/refresh.rs

unifly_api/controller/
refresh.rs
