add ingest analytics page

Cargo.lock

@@ -2530,6 +2530,7 @@ version = "0.1.0"
 dependencies = [
  "async_once",
  "axum",
+ "chrono",
  "futures",
  "futures-util",
  "lazy_static",

services/api/src/analytics.rs

@@ -1,11 +1,50 @@
-use async_graphql::Object;
+use async_graphql::{futures_util::TryStreamExt, Context, Object, SimpleObject};
+use chrono::{DateTime, Utc};
+use sqlx::{query, Pool, Postgres, Row};
 
 pub struct Analytics {}
 
+#[derive(SimpleObject, Debug, Clone)]
+pub struct Event {
+    pub time: DateTime<Utc>,
+    pub event_name: String,
+    pub world_id: i32,
+    pub count: i64,
+}
+
 #[Object]
 impl Analytics {
-    async fn population(&self) -> i32 {
-        0
+    /// Get all events in analytics, bucket_size is in seconds
+    async fn events<'ctx>(
+        &self,
+        ctx: &Context<'ctx>,
+        #[graphql(default = 60)] bucket_size: u64,
+        world_id: Option<i32>,
+    ) -> Vec<Event> {
+        let pool = ctx.data::<Pool<Postgres>>().unwrap();
+
+        let sql = format!("SELECT time_bucket('{} seconds', time) AS bucket, count(*), event_name, world_id FROM analytics WHERE time > now() - interval '1 day' {} GROUP BY bucket, world_id, event_name ORDER BY bucket ASC",
+            bucket_size,
+            if let Some(world_id) = world_id {
+                format!("AND world_id = {}", world_id)
+            } else {
+                "".to_string()
+            }
+        );
+
+        let mut result = query(sql.as_str()).fetch(pool);
+
+        let mut events = Vec::new();
+        while let Some(row) = result.try_next().await.unwrap() {
+            events.push(Event {
+                time: row.get("bucket"),
+                event_name: row.get("event_name"),
+                world_id: row.get("world_id"),
+                count: row.get("count"),
+            });
+        }
+
+        events
     }
 }
 

services/api/src/html/ingest.html

@@ -14,6 +14,213 @@
     color: #cead42;
     text-decoration: none;
   }
+
+  .chart-container {
+    height: 50vh;
+    position: relative;
+  }
+
+  .smaller {
+    height: 33vh;
+  }
 </style>
 <h1>Ingest Stats</h1>
-<p>sorry wip</p>
+<div>
+  <h3>All Events by Type</h3>
+  <div class="chart-container">
+    <canvas id="all-events-by-type" />
+  </div>
+</div>
+<div>
+  <h3>Events by World</h3>
+  <div class="chart-container">
+    <canvas id="events-by-world" />
+  </div>
+</div>
+<div>
+  <h3>Connery</h3>
+  <div class="chart-container smaller">
+    <canvas id="connery" />
+  </div>
+</div>
+<div>
+  <h3>Miller</h3>
+  <div class="chart-container smaller">
+    <canvas id="miller" />
+  </div>
+</div>
+<div>
+  <h3>Cobalt</h3>
+  <div class="chart-container smaller">
+    <canvas id="cobalt" />
+  </div>
+</div>
+<div>
+  <h3>Emerald</h3>
+  <div class="chart-container smaller">
+    <canvas id="emerald" />
+  </div>
+</div>
+<div>
+  <h3>Jaeger</h3>
+  <div class="chart-container smaller">
+    <canvas id="jaeger" />
+  </div>
+</div>
+<div>
+  <h3>SolTech</h3>
+  <div class="chart-container smaller">
+    <canvas id="soltech" />
+  </div>
+</div>
+<div>
+  <h3>Genudine</h3>
+  <div class="chart-container smaller">
+    <canvas id="genudine" />
+  </div>
+</div>
+<div>
+  <h3>Ceres</h3>
+  <div class="chart-container smaller">
+    <canvas id="ceres" />
+  </div>
+</div>
+<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
+<script src="https://cdn.jsdelivr.net/npm/chartjs-adapter-date-fns/dist/chartjs-adapter-date-fns.bundle.min.js"></script>
+<script>
+  const renderChart = (id, data, config = {}) => {};
+
+  const allEventsByType = (id, events) => {
+    let allEvents = events.reduce(
+      (acc, ev) => {
+        acc[ev.eventName][ev.time] = (acc[ev.time] || 0) + ev.count;
+        return acc;
+      },
+      { Death: {}, VehicleDestroy: {} }
+    );
+
+    new Chart(document.getElementById(id), {
+      type: "line",
+      options: {
+        scales: {
+          y: { beginAtZero: true, suggestedMin: 0 },
+          x: { stacked: true, type: "timeseries" },
+        },
+      },
+      data: {
+        datasets: [
+          {
+            label: "Deaths",
+            data: allEvents.Death,
+            // backgroundColor: "#cead42",
+          },
+          {
+            label: "Vehicle Destroys",
+            data: allEvents.VehicleDestroy,
+            // backgroundColor: "#7842ce",
+          },
+        ],
+      },
+    });
+  };
+
+  const eventsByWorld = (events) => {
+    let allEvents = events.reduce((acc, ev) => {
+      acc[ev.worldId] = acc[ev.worldId] || {};
+      acc[ev.worldId][ev.time] = (acc[ev.time] || 0) + ev.count;
+      return acc;
+    }, {});
+
+    new Chart(document.getElementById("events-by-world"), {
+      type: "line",
+      options: {
+        scales: {
+          y: { beginAtZero: true },
+          x: {
+            type: "timeseries",
+          },
+        },
+      },
+      data: {
+        datasets: [
+          {
+            label: "Connery",
+            data: allEvents["1"],
+          },
+          {
+            label: "Miller",
+            data: allEvents["10"],
+          },
+          {
+            label: "Cobalt",
+            data: allEvents["13"],
+          },
+          {
+            label: "Emerald",
+            data: allEvents["17"],
+          },
+          {
+            label: "Jaeger",
+            data: allEvents["19"],
+          },
+          {
+            label: "SolTech",
+            data: allEvents["40"],
+          },
+          {
+            label: "Genudine",
+            data: allEvents["1000"],
+          },
+          {
+            label: "Ceres",
+            data: allEvents["2000"],
+          },
+        ],
+      },
+    });
+  };
+
+  (async () => {
+    let resp = await fetch("/graphql", {
+      method: "POST",
+      headers: {
+        "Content-Type": "application/json",
+      },
+      body: JSON.stringify({
+        query: `
+       {
+        analytics {
+          events(bucketSize: ${60 * 5}) {
+            eventName
+            count
+            time
+            worldId
+          }
+        }
+      }
+    `,
+      }),
+    });
+
+    let body = await resp.json();
+
+    let events = body.data.analytics.events;
+    window.events = events;
+
+    allEventsByType("all-events-by-type", events);
+    eventsByWorld(events);
+    [
+      ["connery", 1],
+      ["miller", 10],
+      ["cobalt", 13],
+      ["emerald", 17],
+      ["jaeger", 19],
+      ["soltech", 40],
+      ["genudine", 1000],
+      ["ceres", 2000],
+    ].forEach(([world, id]) => {
+      let worldEvents = events.filter((ev) => ev.worldId === id);
+      allEventsByType(world, worldEvents);
+    });
+  })();
+</script>

services/api/src/query.rs

@@ -1,6 +1,6 @@
 use crate::{
-    classes::ClassesQuery, health::HealthQuery, population::PopulationQuery,
-    vehicles::VehicleQuery, world::WorldQuery, zone::ZoneQuery,
+    analytics::AnalyticsQuery, classes::ClassesQuery, health::HealthQuery,
+    population::PopulationQuery, vehicles::VehicleQuery, world::WorldQuery, zone::ZoneQuery,
 };
 use async_graphql::MergedObject;
 
@@ -12,4 +12,5 @@ pub struct Query(
     WorldQuery,
     ZoneQuery,
     HealthQuery,
+    AnalyticsQuery,
 );

services/tasks/src/main.rs

@@ -65,6 +65,10 @@ async fn main() {
     match command.as_str() {
         "help" => cmd_help(),
         "prune" => cmd_prune().await,
+        "auto-prune" => loop {
+            cmd_prune().await;
+            tokio::time::sleep(tokio::time::Duration::from_secs(60 * 5)).await;
+        },
         "migrate" => cmd_migrate().await,
         _ => {
             println!("Unknown command: {}", command);

services/websocket/Cargo.toml

@@ -17,4 +17,5 @@ futures-util = "0.3.25"
 futures = "0.3.25"
 async_once = "0.2.6"
 serde-aux = "4.1.2"
-axum = "0.6.1"
+axum = "0.6.1"
+chrono = "0.4.23"

services/websocket/src/main.rs

@@ -1,5 +1,5 @@
 use async_once::AsyncOnce;
-use axum::{routing::get, Router};
+use axum::{routing::get, Json, Router};
 use futures::{pin_mut, FutureExt};
 use futures_util::StreamExt;
 use lazy_static::lazy_static;
@@ -283,7 +283,14 @@ struct Payload {
 }
 
 async fn healthz() {
-    let app = Router::new().route("/healthz", get(|| async { "ok" }));
+    let app = Router::new().route(
+        "/healthz",
+        get(|| async {
+            Json(json!({
+                "status": "ok",
+            }))
+        }),
+    );
 
     let port: u16 = std::env::var("PORT")
         .unwrap_or("8999".to_string())
@@ -314,7 +321,7 @@ async fn main() {
 
     let fused_writer = rx.map(Ok).forward(write).fuse();
     let fused_reader = read
-        .for_each(|msg| async move {
+        .for_each(|msg| async {
             let body = &msg.unwrap().to_string();
 
             let data: Payload = match serde_json::from_str(body) {