xui_rs/events/
models.rs

1use std::{collections::HashMap, fmt};
2
3use serde::{Deserialize, Deserializer, Serialize};
4use serde_json::Value;
5
6use crate::{
7    ActiveInboundsByGuid, ClientTraffic, ClientsByGuid, Error, Inbound, NodeView, OutboundTraffic,
8    ProcessState, Result, ServerStatus,
9};
10
11macro_rules! string_enum {
12    ($name:ident { $($wire:literal => $variant:ident),+ $(,)? }) => {
13        impl $name {
14            /// Returns the exact v3.8.5 wire value.
15            pub fn as_str(&self) -> &str {
16                match self {
17                    $(Self::$variant => $wire,)+
18                    Self::Other(value) => value,
19                }
20            }
21
22            fn from_wire(value: String) -> Self {
23                match value.as_str() {
24                    $($wire => Self::$variant,)+
25                    _ => Self::Other(value),
26                }
27            }
28        }
29
30        impl Serialize for $name {
31            fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
32            where
33                S: serde::Serializer,
34            {
35                serializer.serialize_str(self.as_str())
36            }
37        }
38
39        impl<'de> Deserialize<'de> for $name {
40            fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
41            where
42                D: Deserializer<'de>,
43            {
44                String::deserialize(deserializer).map(Self::from_wire)
45            }
46        }
47    };
48}
49
50/// Message names declared by the v3.8.5 WebSocket hub.
51#[derive(Clone, Debug, Eq, Hash, PartialEq)]
52#[non_exhaustive]
53pub enum EventMessageType {
54    /// Complete machine and Xray status snapshot.
55    Status,
56    /// Live local/node traffic update.
57    Traffic,
58    /// Complete inbound-list replacement.
59    Inbounds,
60    /// Complete outbound-traffic replacement.
61    Outbounds,
62    /// Complete node-list replacement.
63    Nodes,
64    /// In-panel operator notification.
65    Notification,
66    /// Xray process-state transition.
67    XrayState,
68    /// Absolute client and inbound traffic counters.
69    ClientStats,
70    /// Reserved client-collection message name.
71    Clients,
72    /// Instruction to refresh a resource through HTTP.
73    Invalidate,
74    /// Message introduced by a newer panel version.
75    Other(String),
76}
77
78string_enum!(EventMessageType {
79    "status" => Status,
80    "traffic" => Traffic,
81    "inbounds" => Inbounds,
82    "outbounds" => Outbounds,
83    "nodes" => Nodes,
84    "notification" => Notification,
85    "xray_state" => XrayState,
86    "client_stats" => ClientStats,
87    "clients" => Clients,
88    "invalidate" => Invalidate,
89});
90
91/// Severity attached to a panel notification.
92#[derive(Clone, Debug, Eq, Hash, PartialEq)]
93#[non_exhaustive]
94pub enum NotificationLevel {
95    /// Successful operation.
96    Success,
97    /// Informational notice.
98    Info,
99    /// Warning that may require attention.
100    Warning,
101    /// Failed operation or unhealthy state.
102    Error,
103    /// Level introduced by a newer panel version.
104    Other(String),
105}
106
107string_enum!(NotificationLevel {
108    "success" => Success,
109    "info" => Info,
110    "warning" => Warning,
111    "error" => Error,
112});
113
114/// One Xray traffic delta. Its capitalized field names are the exact Go wire
115/// representation used by v3.8.5.
116#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
117#[serde(default, rename_all = "PascalCase")]
118pub struct TrafficDelta {
119    /// Whether the tag identifies an inbound.
120    pub is_inbound: bool,
121    /// Whether the tag identifies an outbound.
122    pub is_outbound: bool,
123    /// Stable Xray tag.
124    pub tag: String,
125    /// Uploaded bytes during the collection window.
126    pub up: i64,
127    /// Downloaded bytes during the collection window.
128    pub down: i64,
129}
130
131/// Partial live-traffic payload emitted independently by the local Xray and
132/// remote-node synchronization jobs.
133#[derive(Clone, Default, Deserialize, PartialEq, Serialize)]
134#[serde(default, rename_all = "camelCase")]
135pub struct TrafficUpdate {
136    /// Local Xray inbound/outbound deltas, when this is a local poll.
137    pub traffics: Option<Vec<TrafficDelta>>,
138    /// Remote-node inbound deltas, when this is a node synchronization poll.
139    pub node_traffics: Option<Vec<TrafficDelta>>,
140    /// Per-client deltas that moved bytes in the latest local poll.
141    pub client_traffics: Option<Vec<ClientTraffic>>,
142    /// Globally online client emails.
143    #[serde(deserialize_with = "deserialize_null_default")]
144    pub online_clients: Vec<String>,
145    /// Online client emails grouped by panel GUID.
146    #[serde(deserialize_with = "deserialize_null_default")]
147    pub online_by_guid: ClientsByGuid,
148    /// Active inbound tags grouped by panel GUID.
149    #[serde(deserialize_with = "deserialize_null_default")]
150    pub active_inbounds: ActiveInboundsByGuid,
151    /// Last-online timestamps in milliseconds keyed by client email.
152    #[serde(deserialize_with = "deserialize_null_default")]
153    pub last_online_map: HashMap<String, i64>,
154}
155
156impl fmt::Debug for TrafficUpdate {
157    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
158        formatter
159            .debug_struct("TrafficUpdate")
160            .field("traffics", &self.traffics.as_ref().map(Vec::len))
161            .field("node_traffics", &self.node_traffics.as_ref().map(Vec::len))
162            .field(
163                "client_traffics",
164                &self.client_traffics.as_ref().map(Vec::len),
165            )
166            .field("online_clients", &self.online_clients.len())
167            .field("online_by_guid", &self.online_by_guid.len())
168            .field("active_inbounds", &self.active_inbounds.len())
169            .field("last_online_map", &self.last_online_map.len())
170            .finish()
171    }
172}
173
174/// Absolute traffic summary for one inbound.
175#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
176#[serde(default)]
177pub struct InboundTrafficSummary {
178    /// Inbound database identifier.
179    pub id: i64,
180    /// Cumulative uploaded bytes.
181    pub up: i64,
182    /// Cumulative downloaded bytes.
183    pub down: i64,
184    /// Traffic allowance in bytes.
185    pub total: i64,
186    /// Whether the inbound is enabled.
187    pub enable: bool,
188}
189
190/// Absolute client/inbound counters emitted after a traffic collection.
191#[derive(Clone, Default, Deserialize, PartialEq, Serialize)]
192#[serde(default, rename_all = "camelCase")]
193pub struct ClientStatsUpdate {
194    /// `true` when `clients` replaces the complete client-stat collection;
195    /// `false` when it contains only active rows.
196    pub snapshot: bool,
197    /// Absolute per-client counters.
198    #[serde(deserialize_with = "deserialize_null_default")]
199    pub clients: Vec<ClientTraffic>,
200    /// Absolute per-inbound summaries.
201    #[serde(deserialize_with = "deserialize_null_default")]
202    pub inbounds: Vec<InboundTrafficSummary>,
203}
204
205impl fmt::Debug for ClientStatsUpdate {
206    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
207        formatter
208            .debug_struct("ClientStatsUpdate")
209            .field("snapshot", &self.snapshot)
210            .field("clients", &self.clients.len())
211            .field("inbounds", &self.inbounds.len())
212            .finish()
213    }
214}
215
216/// Operator notification pushed by 3x-ui.
217#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
218#[serde(rename_all = "camelCase")]
219pub struct PanelNotification {
220    /// Localized notification title.
221    pub title: String,
222    /// Notification body.
223    pub message: String,
224    /// Notification severity.
225    pub level: NotificationLevel,
226}
227
228impl fmt::Debug for PanelNotification {
229    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
230        formatter
231            .debug_struct("PanelNotification")
232            .field("title", &"[REDACTED]")
233            .field("message", &"[REDACTED]")
234            .field("level", &self.level)
235            .finish()
236    }
237}
238
239/// Xray process-state transition pushed after lifecycle operations.
240#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
241#[serde(rename_all = "camelCase")]
242pub struct XrayStateChange {
243    /// New process state.
244    pub state: ProcessState,
245    /// Startup/reload error, when `state` is `error`.
246    pub error_msg: String,
247}
248
249impl fmt::Debug for XrayStateChange {
250    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
251        formatter
252            .debug_struct("XrayStateChange")
253            .field("state", &self.state)
254            .field("error_msg", &"[REDACTED]")
255            .finish()
256    }
257}
258
259/// Lightweight request to refresh one resource through its typed HTTP API.
260#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
261pub struct Invalidation {
262    /// Original data type that could not or should not be pushed directly.
263    #[serde(rename = "type")]
264    pub target: EventMessageType,
265}
266
267/// Decoded payload of one v3.8.5 real-time message.
268#[derive(Clone, PartialEq)]
269#[non_exhaustive]
270pub enum PanelEventKind {
271    /// Complete server-status snapshot.
272    Status(Box<ServerStatus>),
273    /// Partial live traffic update.
274    Traffic(Box<TrafficUpdate>),
275    /// Complete inbound-list replacement.
276    Inbounds(Vec<Inbound>),
277    /// Complete outbound-traffic replacement.
278    Outbounds(Vec<OutboundTraffic>),
279    /// Complete node-tree replacement.
280    Nodes(Vec<NodeView>),
281    /// Operator notification.
282    Notification(PanelNotification),
283    /// Xray process transition.
284    XrayState(XrayStateChange),
285    /// Absolute client/inbound counters.
286    ClientStats(ClientStatsUpdate),
287    /// Reserved source-declared clients message. v3.8.5 has no direct
288    /// broadcaster or stable payload schema for this type.
289    Clients(Value),
290    /// Request to refresh a resource through HTTP.
291    Invalidate(Invalidation),
292    /// Message introduced by a newer panel version. Its payload is preserved.
293    Unknown {
294        /// Exact unrecognized message name.
295        message_type: String,
296        /// Exact unrecognized JSON payload.
297        payload: Value,
298    },
299}
300
301impl PanelEventKind {
302    /// Returns the source-defined or forward-compatible message name.
303    pub fn message_type(&self) -> EventMessageType {
304        match self {
305            Self::Status(_) => EventMessageType::Status,
306            Self::Traffic(_) => EventMessageType::Traffic,
307            Self::Inbounds(_) => EventMessageType::Inbounds,
308            Self::Outbounds(_) => EventMessageType::Outbounds,
309            Self::Nodes(_) => EventMessageType::Nodes,
310            Self::Notification(_) => EventMessageType::Notification,
311            Self::XrayState(_) => EventMessageType::XrayState,
312            Self::ClientStats(_) => EventMessageType::ClientStats,
313            Self::Clients(_) => EventMessageType::Clients,
314            Self::Invalidate(_) => EventMessageType::Invalidate,
315            Self::Unknown { message_type, .. } => EventMessageType::Other(message_type.clone()),
316        }
317    }
318}
319
320impl fmt::Debug for PanelEventKind {
321    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
322        match self {
323            Self::Status(status) => formatter
324                .debug_tuple("Status")
325                .field(&status.xray.state)
326                .finish(),
327            Self::Traffic(update) => formatter.debug_tuple("Traffic").field(update).finish(),
328            Self::Inbounds(rows) => formatter
329                .debug_struct("Inbounds")
330                .field("count", &rows.len())
331                .finish(),
332            Self::Outbounds(rows) => formatter
333                .debug_struct("Outbounds")
334                .field("count", &rows.len())
335                .finish(),
336            Self::Nodes(rows) => formatter
337                .debug_struct("Nodes")
338                .field("count", &rows.len())
339                .finish(),
340            Self::Notification(value) => {
341                formatter.debug_tuple("Notification").field(value).finish()
342            }
343            Self::XrayState(value) => formatter.debug_tuple("XrayState").field(value).finish(),
344            Self::ClientStats(value) => formatter.debug_tuple("ClientStats").field(value).finish(),
345            Self::Clients(_) => formatter.write_str("Clients([REDACTED])"),
346            Self::Invalidate(value) => formatter.debug_tuple("Invalidate").field(value).finish(),
347            Self::Unknown { message_type, .. } => formatter
348                .debug_struct("Unknown")
349                .field("message_type", message_type)
350                .field("payload", &"[REDACTED]")
351                .finish(),
352        }
353    }
354}
355
356/// One typed real-time event from the panel.
357#[derive(Clone, Debug, PartialEq)]
358pub struct PanelEvent {
359    /// Server-generated Unix timestamp in milliseconds.
360    pub timestamp_ms: i64,
361    /// Typed message payload.
362    pub kind: PanelEventKind,
363}
364
365#[derive(Deserialize)]
366struct RawEvent {
367    #[serde(rename = "type")]
368    message_type: String,
369    payload: Value,
370    time: i64,
371}
372
373impl PanelEvent {
374    pub(crate) fn decode(text: &str) -> Result<Self> {
375        let raw: RawEvent = serde_json::from_str(text).map_err(|source| Error::EventDecode {
376            message_type: None,
377            source,
378        })?;
379        let message_type = EventMessageType::from_wire(raw.message_type.clone());
380        let payload = raw.payload;
381        let kind = match message_type {
382            EventMessageType::Status => {
383                PanelEventKind::Status(Box::new(decode_payload(&raw.message_type, payload)?))
384            }
385            EventMessageType::Traffic => {
386                PanelEventKind::Traffic(Box::new(decode_payload(&raw.message_type, payload)?))
387            }
388            EventMessageType::Inbounds => {
389                PanelEventKind::Inbounds(decode_payload(&raw.message_type, payload)?)
390            }
391            EventMessageType::Outbounds => {
392                PanelEventKind::Outbounds(decode_payload(&raw.message_type, payload)?)
393            }
394            EventMessageType::Nodes => {
395                PanelEventKind::Nodes(decode_payload(&raw.message_type, payload)?)
396            }
397            EventMessageType::Notification => {
398                PanelEventKind::Notification(decode_payload(&raw.message_type, payload)?)
399            }
400            EventMessageType::XrayState => {
401                PanelEventKind::XrayState(decode_payload(&raw.message_type, payload)?)
402            }
403            EventMessageType::ClientStats => {
404                PanelEventKind::ClientStats(decode_payload(&raw.message_type, payload)?)
405            }
406            EventMessageType::Clients => PanelEventKind::Clients(payload),
407            EventMessageType::Invalidate => {
408                PanelEventKind::Invalidate(decode_payload(&raw.message_type, payload)?)
409            }
410            EventMessageType::Other(message_type) => PanelEventKind::Unknown {
411                message_type,
412                payload,
413            },
414        };
415        Ok(Self {
416            timestamp_ms: raw.time,
417            kind,
418        })
419    }
420}
421
422fn decode_payload<T>(message_type: &str, payload: Value) -> Result<T>
423where
424    T: for<'de> Deserialize<'de>,
425{
426    serde_json::from_value(payload).map_err(|source| Error::EventDecode {
427        message_type: Some(message_type.to_owned()),
428        source,
429    })
430}
431
432fn deserialize_null_default<'de, D, T>(deserializer: D) -> std::result::Result<T, D::Error>
433where
434    D: Deserializer<'de>,
435    T: Deserialize<'de> + Default,
436{
437    Option::<T>::deserialize(deserializer).map(Option::unwrap_or_default)
438}
439
440#[cfg(test)]
441mod tests {
442    use super::*;
443    use serde_json::json;
444
445    #[test]
446    fn extensible_string_enums_round_trip_every_wire_value() {
447        let message_types = [
448            EventMessageType::Status,
449            EventMessageType::Traffic,
450            EventMessageType::Inbounds,
451            EventMessageType::Outbounds,
452            EventMessageType::Nodes,
453            EventMessageType::Notification,
454            EventMessageType::XrayState,
455            EventMessageType::ClientStats,
456            EventMessageType::Clients,
457            EventMessageType::Invalidate,
458            EventMessageType::Other("future_event".into()),
459        ];
460        for value in message_types {
461            let wire = value.as_str().to_owned();
462            assert_eq!(serde_json::to_value(&value).unwrap(), wire);
463            assert_eq!(
464                serde_json::from_value::<EventMessageType>(json!(wire)).unwrap(),
465                value
466            );
467        }
468
469        let levels = [
470            NotificationLevel::Success,
471            NotificationLevel::Info,
472            NotificationLevel::Warning,
473            NotificationLevel::Error,
474            NotificationLevel::Other("critical".into()),
475        ];
476        for value in levels {
477            let wire = value.as_str().to_owned();
478            assert_eq!(serde_json::to_value(&value).unwrap(), wire);
479            assert_eq!(
480                serde_json::from_value::<NotificationLevel>(json!(wire)).unwrap(),
481                value
482            );
483        }
484    }
485
486    #[test]
487    fn event_debug_views_redact_payloads_and_report_shape() {
488        let traffic = TrafficUpdate {
489            traffics: Some(vec![TrafficDelta::default()]),
490            node_traffics: Some(vec![]),
491            online_clients: vec!["private@example.com".into()],
492            ..TrafficUpdate::default()
493        };
494        let stats = ClientStatsUpdate {
495            snapshot: true,
496            inbounds: vec![InboundTrafficSummary::default()],
497            ..ClientStatsUpdate::default()
498        };
499        let notification = PanelNotification {
500            title: "private-title".into(),
501            message: "private-message".into(),
502            level: NotificationLevel::Warning,
503        };
504        let xray_state = XrayStateChange {
505            state: ProcessState::Error,
506            error_msg: "private-xray-error".into(),
507        };
508
509        assert!(!format!("{traffic:?}").contains("private@example.com"));
510        assert!(format!("{stats:?}").contains("snapshot: true"));
511        assert!(!format!("{notification:?}").contains("private-title"));
512        assert!(!format!("{notification:?}").contains("private-message"));
513        assert!(!format!("{xray_state:?}").contains("private-xray-error"));
514    }
515
516    #[test]
517    fn every_event_kind_reports_its_type_without_exposing_payloads() {
518        let values = vec![
519            PanelEventKind::Status(Box::default()),
520            PanelEventKind::Traffic(Box::default()),
521            PanelEventKind::Inbounds(Vec::new()),
522            PanelEventKind::Outbounds(Vec::new()),
523            PanelEventKind::Nodes(Vec::new()),
524            PanelEventKind::Notification(PanelNotification {
525                title: "notification-secret".into(),
526                message: "message-secret".into(),
527                level: NotificationLevel::Info,
528            }),
529            PanelEventKind::XrayState(XrayStateChange {
530                state: ProcessState::Running,
531                error_msg: "state-secret".into(),
532            }),
533            PanelEventKind::ClientStats(ClientStatsUpdate::default()),
534            PanelEventKind::Clients(json!({"password": "clients-secret"})),
535            PanelEventKind::Invalidate(Invalidation {
536                target: EventMessageType::Inbounds,
537            }),
538            PanelEventKind::Unknown {
539                message_type: "future".into(),
540                payload: json!({"password": "future-secret"}),
541            },
542        ];
543        let expected = [
544            EventMessageType::Status,
545            EventMessageType::Traffic,
546            EventMessageType::Inbounds,
547            EventMessageType::Outbounds,
548            EventMessageType::Nodes,
549            EventMessageType::Notification,
550            EventMessageType::XrayState,
551            EventMessageType::ClientStats,
552            EventMessageType::Clients,
553            EventMessageType::Invalidate,
554            EventMessageType::Other("future".into()),
555        ];
556        for (value, expected) in values.into_iter().zip(expected) {
557            assert_eq!(value.message_type(), expected);
558            let output = format!("{value:?}");
559            for secret in [
560                "notification-secret",
561                "message-secret",
562                "state-secret",
563                "clients-secret",
564                "future-secret",
565            ] {
566                assert!(!output.contains(secret));
567            }
568        }
569    }
570
571    #[test]
572    fn malformed_envelope_has_no_claimed_message_type() {
573        let error = PanelEvent::decode("not-json").unwrap_err();
574        assert!(matches!(
575            error,
576            Error::EventDecode {
577                message_type: None,
578                ..
579            }
580        ));
581    }
582}
xui_rs/events/models.rs

xui_rs/events/
models.rs
