feat: add analytics max concurrent listeners (#116)

defs/analytics/Analytics.ts

@@ -14,6 +14,7 @@ export type Analytics = {
   ips: number;
   total_duration_ms: number;
   total_transfer_bytes: number;
+  max_concurrent_listeners: number;
   by_month: Array<AnalyticsItem<YearMonth>>;
   by_day: Array<AnalyticsItem<YearMonthDay>>;
   by_hour: Array<AnalyticsItem<number>>;

defs/analytics/AnalyticsItem.ts

@@ -6,4 +6,5 @@ export type AnalyticsItem<K> = {
   ips: number;
   total_duration_ms: number;
   total_transfer_bytes: number;
+  max_concurrent_listeners: number;
 };

front/server/src/locale/share/analytics/analytics.ar.ts

@@ -8,6 +8,7 @@ const locale: import("./analytics.locale").AnalyticsLocale = {
   "Total_transfer": "إجمالي النقل",
   "Total_transfer_in_MB": "إجمالي النقل بالميجابايت",
   "Unique_IPs": "عناوين الـ IP الفريدة",
+  "Max_concurrent_listeners": "الحد الأقصى للمستمعين المتزامنين",
 
   // @example Browser: Chrome
   "Browser": "المتصفح",

front/server/src/locale/share/analytics/analytics.de.ts

@@ -8,6 +8,7 @@ const locale: import("./analytics.locale").AnalyticsLocale = {
   "Total_transfer": "Gesamtübertragung",
   "Total_transfer_in_MB": "Gesamtübertragung in MB",
   "Unique_IPs": "Einzigartige IPs",
+  "Max_concurrent_listeners": "Max. gleichzeitige Zuhörer",
 
   // @example Browser: Chrome
   "Browser": "Browser",

front/server/src/locale/share/analytics/analytics.en.ts

@@ -7,6 +7,7 @@ const locale = {
   "Total_transfer": "Total transfer",
   "Total_transfer_in_MB": "Total transfer in MB",
   "Unique_IPs": "Unique IPs",
+  "Max_concurrent_listeners": "Max. concurrent listeners",
 
   // @example Browser: Chrome
   "Browser": "Browser",

front/server/src/locale/share/analytics/analytics.es.ts

@@ -7,6 +7,7 @@ const locale: import("./analytics.locale").AnalyticsLocale = {
   "Total_transfer": "Transferencia total",
   "Total_transfer_in_MB": "Transferencia total en MB",
   "Unique_IPs": "IPs únicas",
+  "Max_concurrent_listeners": "Máx. oyentes simultáneos",
 
   // @example Browser: Chrome
   "Browser": "Navegador",

front/server/src/locale/share/analytics/analytics.fr.ts

@@ -8,6 +8,7 @@ const locale: import("./analytics.locale").AnalyticsLocale = {
   "Total_transfer": "Transfert total",
   "Total_transfer_in_MB": "Transfert total en MB",
   "Unique_IPs": "IPs uniques",
+  "Max_concurrent_listeners": "Max. auditeurs simultanés",
 
   // @example Browser: Chrome
   "Browser": "Navigateur",

front/server/src/locale/share/analytics/analytics.it.ts

@@ -8,7 +8,8 @@ const locale: import("./analytics.locale").AnalyticsLocale = {
   "Total_transfer": "Trasferimento totale",
   "Total_transfer_in_MB": "Trasferimento totale in MB",
   "Unique_IPs": "IP unici",
-
+  "Max_concurrent_listeners": "Max. ascoltatori simultanei",
+
   // @example Browser: Chrome
   "Browser": "Browser",
 

front/server/src/locale/share/analytics/analytics.pt.ts

@@ -8,6 +8,7 @@ const locale: import("./analytics.locale").AnalyticsLocale = {
   "Total_transfer": "Transferência total",
   "Total_transfer_in_MB": "Transferência total em MB",
   "Unique_IPs": "IPs únicos",
+  "Max_concurrent_listeners": "Máx. ouvintes simultâneos",
 
   // @example Browser: Chrome
   "Browser": "Navegador",

front/server/src/locale/share/analytics/analytics.zh.ts

@@ -8,6 +8,7 @@ const locale: import("./analytics.locale").AnalyticsLocale = {
   "Total_transfer": "总传输量",
   "Total_transfer_in_MB": "总传输量（MB）",
   "Unique_IPs": "唯一IP",
+  "Max_concurrent_listeners": "最大并发监听者",
 
   // @example Browser: Chrome
   "Browser": "浏览器",

front/share/src/analytics/AnalyticsData.svelte

@@ -514,7 +514,13 @@
 
   const get_common_grid_options = () => {
     
-    type Item = {  sessions: number, ips: number, total_duration_ms: number, total_transfer_bytes: number }
+    type Item = {
+      sessions: number,
+      ips: number,
+      total_duration_ms: number,
+      total_transfer_bytes: number,
+      max_concurrent_listeners: number
+    }
 
     const fields = {
       "sessions": {
@@ -531,6 +537,13 @@
         numeric: true,
       },
 
+      "max_concurrent": {
+        name: locale.Max_concurrent_listeners,
+        format: item => to_fixed(item.max_concurrent_listeners, 0),
+        sort: (a, b) => compare_numbers(a.max_concurrent_listeners, b.max_concurrent_listeners),
+        numeric: true,
+      },
+
       "avg_time": {
         name: locale.Average_listening_minutes,
         format: item => item.sessions === 0 ? "-" : format_mins(item.total_duration_ms / item.sessions),
@@ -718,6 +731,7 @@
           ips: 0,
           total_duration_ms: 0,
           total_transfer_bytes: 0,
+          max_concurrent_listeners: 0,
         })
       } else {
         items.push(item)
@@ -915,6 +929,13 @@
         </div>
       </div>
 
+      <div class="total">
+        <div class="total-title">{locale.Max_concurrent_listeners}</div>
+        <div class="total-value">
+          {data.max_concurrent_listeners}
+        </div>
+      </div>
+
       <div class="total">
         <div class="total-title">{locale.Average_listening_time}</div>
         <div class="total-value">

rs/packages/db/src/models/stream_connection/analytics/mod.rs

@@ -40,6 +40,9 @@ pub struct Analytics {
   #[serde(with = "serde_util::as_f64")]
   pub total_transfer_bytes: u64,
 
+  #[serde(with = "serde_util::as_f64")]
+  pub max_concurrent_listeners: u64,
+
   pub by_month: Vec<AnalyticsItem<YearMonth>>,
   pub by_day: Vec<AnalyticsItem<YearMonthDay>>,
   pub by_hour: Vec<AnalyticsItem<u8>>,
@@ -62,6 +65,8 @@ pub struct AnalyticsItem<K> {
   total_duration_ms: u64,
   #[serde(with = "serde_util::as_f64")]
   total_transfer_bytes: u64,
+  #[serde(with = "serde_util::as_f64")]
+  max_concurrent_listeners: u64,
 }
 
 #[derive(Debug, Clone, Copy, Serialize, Eq, PartialEq, Hash, Ord, PartialOrd, Deserialize, TS)]
@@ -236,12 +241,16 @@ pub async fn get_analytics(query: AnalyticsQuery) -> Result<Analytics, mongodb::
   let mut total_duration_ms: u64 = 0;
   let mut total_transfer_bytes: u64 = 0;
 
+  // u32 is the timestamp and bool is true => start, false => stop
+  let mut start_stop_events: Vec<(u32, bool)> = vec![];
+
   #[derive(Default)]
   struct AccumulatorItem {
     sessions: u64,
     ips: HashSet<IpAddr>,
     total_duration_ms: u64,
     total_transfer_bytes: u64,
+    start_stop_events: Vec<(u32, bool)>,
   }
 
   let mut months_accumulator = HashMap::<YearMonth, AccumulatorItem>::new();
@@ -265,10 +274,15 @@ pub async fn get_analytics(query: AnalyticsQuery) -> Result<Analytics, mongodb::
     let conn_browser = conn.browser;
     let conn_os = conn.os;
 
+    let start = created_at.unix_timestamp() as u32;
+    let stop = start + (conn_duration_ms / 1000) as u32;
+
     sessions += 1;
     total_duration_ms += conn_duration_ms;
     total_transfer_bytes += conn_transfer_bytes;
     ips.insert(conn.ip);
+    start_stop_events.push((start, true));
+    start_stop_events.push((stop, false));
 
     macro_rules! add {
       ($acc:ident, $key:expr) => {
@@ -277,6 +291,8 @@ pub async fn get_analytics(query: AnalyticsQuery) -> Result<Analytics, mongodb::
         item.ips.insert(conn.ip);
         item.total_duration_ms += conn_duration_ms;
         item.total_transfer_bytes += conn_transfer_bytes;
+        item.start_stop_events.push((start, true));
+        item.start_stop_events.push((stop, false));
       };
     }
 
@@ -303,6 +319,29 @@ pub async fn get_analytics(query: AnalyticsQuery) -> Result<Analytics, mongodb::
     add!(domain_accumulator, conn.domain);
   }
 
+  macro_rules! max_concurrent {
+    ($vec:expr) => {{
+      let mut vec = $vec;
+      vec.sort_by(|a, b| a.0.cmp(&b.0));
+
+      let mut max: u32 = 0;
+      let mut current: u32 = 0;
+      for (_, start) in vec.into_iter() {
+        if start {
+          current = current.saturating_add(1);
+        } else {
+          current = current.saturating_sub(1);
+        }
+
+        if current > max {
+          max = current;
+        }
+      }
+
+      max as u64
+    }};
+  }
+
   macro_rules! collect {
     ($acc:ident) => {
       $acc
@@ -313,6 +352,7 @@ pub async fn get_analytics(query: AnalyticsQuery) -> Result<Analytics, mongodb::
           ips: value.ips.len() as u64,
           total_duration_ms: value.total_duration_ms,
           total_transfer_bytes: value.total_transfer_bytes,
+          max_concurrent_listeners: max_concurrent!(value.start_stop_events),
         })
         .collect::<Vec<_>>()
     };
@@ -360,6 +400,7 @@ pub async fn get_analytics(query: AnalyticsQuery) -> Result<Analytics, mongodb::
     total_duration_ms,
     total_transfer_bytes,
     ips: ips.len() as u64,
+    max_concurrent_listeners: max_concurrent!(start_stop_events),
     stations,
     by_month,
     by_day,