+
{move || Suspend::new(async move {
let configs = match configs.await {
@@ -249,7 +368,87 @@ pub fn InstancesView() -> impl IntoView {
/>
})
})}
+
+
+
+
+
display_checkins.get(),
+ _ => true
+ })
+ key=|lineitem| lineitem
+ .as_ref()
+ .map(move |li| li.date())
+ .map_err(Clone::clone)
+ let:lineitem
+ >
+ {match lineitem {
+ Ok(LineItem::Checkin(date)) => EitherOf5::A(view! {
+
+ "Beacon checked in at " {format!("{}", date)}
+
+ }),
+ Ok(LineItem::CommandResult {
+ result_date,
+ action,
+ action_result,
+ command_id,
+ ..
+ }) => EitherOf5::B(view! {
+
+
+
+ {action.render_data(action_result)}
+
+
+
+
+ {action.render_empty()}
+
+
+ "There was an issue decoding the command result in the database"
+
+ }),
+ Err(InvokeOrResult::Invoke) => EitherOf5::E(view! {
+
+ "There was an issue decoding the command sent in the database"
+
+ })
+ }}
+
+
+ {move || done_with_scrolling.get().then(|| view! {
+
+
+ "No more history entries exist for this beacon"
+
+
+ })}
+
}
}
diff --git a/sparse-server/src/beacons/sidebar.rs b/sparse-server/src/beacons/sidebar.rs
index 14a27b2..87f305e 100644
--- a/sparse-server/src/beacons/sidebar.rs
+++ b/sparse-server/src/beacons/sidebar.rs
@@ -2,7 +2,7 @@ use std::sync::Arc;
use leptos::{either::Either, prelude::*};
use leptos_router::components::A;
-#[cfg(feature = "hydrate")]
+#[cfg(not(feature = "ssr"))]
use leptos_use::use_websocket;
use serde::{Deserialize, Serialize};
@@ -599,30 +599,28 @@ pub fn BeaconSidebar() -> impl IntoView {
"Version: " {beacon.version.to_string()}
- {(sort_method.get() != Some(SortMethod::Category))
- .then(|| -> Vec
{
- let BeaconResources {
- categories,
- ..
- } = expect_context::();
+ {Some((|| {
+ let BeaconResources {
+ categories,
+ ..
+ } = expect_context::();
- let Some(Ok(ref categories)) = *categories.read() else {
- return vec![];
- };
+ let Some(Ok(ref categories)) = *categories.read() else {
+ return vec![];
+ };
- categories
- .iter()
- .filter(|cat| beacon.category_ids.contains(&cat.category_id))
- .map(|cat| cat.category_name.clone())
- .collect()
- })
+ categories
+ .iter()
+ .filter(|cat| beacon.category_ids.contains(&cat.category_id))
+ .map(|cat| cat.category_name.clone())
+ .collect::>()
+ })())
.filter(|categories| !categories.is_empty())
.map(|categories| view! {
"Categories: " {categories.join(", ")}
- })
- }
+ })}
{(sort_method.get() != Some(SortMethod::Template))
.then(|| -> Option {
let BeaconResources {
diff --git a/sparse-server/src/webserver.rs b/sparse-server/src/webserver.rs
index eaaf6e7..48f3832 100644
--- a/sparse-server/src/webserver.rs
+++ b/sparse-server/src/webserver.rs
@@ -532,6 +532,7 @@ async fn handle_listener_events(
pub async fn attach_to_beacon(
State(state): State,
cookie_jar: CookieJar,
+ Path(beacon_id): Path,
ws: ws::WebSocketUpgrade
) -> axum::response::Response {
let user = match crate::db::user::get_auth_session_inner(
@@ -551,7 +552,7 @@ pub async fn attach_to_beacon(
ws
.on_upgrade(move |socket: ws::WebSocket| async move {
- if let Err(e) = handle_beacon_socket(socket, state).await {
+ if let Err(e) = handle_beacon_socket(beacon_id, socket, state).await {
tracing::warn!("Encountered error when handling beacon subscriber: {e}");
};
})
@@ -559,12 +560,237 @@ pub async fn attach_to_beacon(
}
async fn handle_beacon_socket(
+ beacon_id: String,
mut socket: ws::WebSocket,
state: AppState,
) -> Result<(), crate::error::Error> {
- unimplemented!()
-}
+ let mut event_receiver = state.beacon_event_broadcast.subscribe();
+ use crate::beacons::instances::{BeaconClientMessage, BeaconViewEvent, InvokeOrResult, LineItem};
+
+ macro_rules! send_event {
+ ($ev:expr) => {{
+ let ev = $ev;
+ let json = serde_json::to_string(&ev)?;
+ socket.send(ws::Message::Text(json)).await?;
+ }}
+ }
+
+ loop {
+ use sparse_handler::BeaconEvent;
+
+ tokio::select! {
+ event = event_receiver.recv() => match event {
+ Ok(BeaconEvent::BeaconCommandFinished(bid, cid)) if bid == beacon_id => {
+ struct InvokeInfo {
+ params: String,
+ date: chrono::DateTime,
+ res: String,
+ user_name: String
+ }
+
+ type DateTime = chrono::DateTime;
+
+ // Null safety: the only time the above event is published is after
+ // the database has updated the two values
+ // It would maybe wise to update the event to instead pass the results...
+ let invoke_info = sqlx::query_as!(
+ InvokeInfo,
+ r#"SELECT cmd.cmd_parameters as params, bci.invocation_result as "res!",
+ bci.invocation_date as "date!: DateTime", users.user_name as user_name
+ FROM beacon_command_invocation bci
+ INNER JOIN users ON users.user_id = bci.invoker_id
+ INNER JOIN beacon_command cmd ON cmd.command_id = bci.command_id
+ WHERE bci.command_id = ? AND bci.beacon_id = ?"#,
+ cid,
+ bid
+ )
+ .fetch_one(&state.db)
+ .await?;
+
+ send_event!(BeaconViewEvent::NewEvent(LineItem::CommandResult {
+ user_name: invoke_info.user_name,
+ result_date: invoke_info.date,
+ command_id: cid,
+ action: serde_json::from_str(&invoke_info.params)?,
+ action_result: invoke_info.res
+ }));
+ }
+ Ok(BeaconEvent::BeaconUpdate(bid)) if bid == beacon_id => {
+ send_event!(BeaconViewEvent::BeaconUpdate);
+ }
+ Ok(BeaconEvent::Checkin(bid)) if bid == beacon_id => {
+ send_event!(BeaconViewEvent::NewEvent(LineItem::Checkin(chrono::Utc::now())));
+ }
+ Ok(BeaconEvent::CommandIssued(bid, cid)) if bid == beacon_id => {
+ struct InvokeInfo {
+ params: String,
+ date: chrono::DateTime,
+ user_name: String
+ }
+
+ type DateTime = chrono::DateTime;
+
+ // Null safety: the only time the above event is published is after
+ // the database has updated the two values
+ // It would maybe wise to update the event to instead pass the results...
+ let invoke_info = sqlx::query_as!(
+ InvokeInfo,
+ r#"SELECT cmd.cmd_parameters as params, bci.issue_date as "date: DateTime",
+ users.user_name as user_name
+ FROM beacon_command_invocation bci
+ INNER JOIN users ON users.user_id = bci.invoker_id
+ INNER JOIN beacon_command cmd ON cmd.command_id = bci.command_id
+ WHERE bci.command_id = ? AND bci.beacon_id = ?"#,
+ cid,
+ bid
+ )
+ .fetch_one(&state.db)
+ .await?;
+
+ send_event!(BeaconViewEvent::NewEvent(LineItem::CommandInvocation {
+ user_name: invoke_info.user_name,
+ invoke_date: invoke_info.date,
+ command_id: cid,
+ action: serde_json::from_str(&invoke_info.params)?,
+ }));
+ }
+ Ok(_) => {
+ // not for us
+ }
+ Err(e) => {
+ tracing::warn!("Unable to handle general event: {e:?}");
+ }
+ },
+ user_event = socket.recv() => match user_event {
+ Some(Ok(ws::Message::Text(ev))) => {
+ let Ok(msg) = serde_json::from_str::(&ev) else {
+ continue;
+ };
+
+ let BeaconClientMessage::LoadHistorical(offset) = msg;
+
+ #[derive(Debug)]
+ enum DbLineItem {
+ CommandInvocation {
+ user_name: String,
+ invoke_date: chrono::DateTime,
+ command_id: i64,
+ action: String
+ },
+ CommandResult {
+ user_name: String,
+ result_date: chrono::DateTime,
+ command_id: i64,
+ action: String,
+ action_result: String
+ },
+ Checkin(chrono::DateTime)
+ }
+
+ use sqlx::{FromRow, sqlite::SqliteRow, Row};
+ impl FromRow<'_, SqliteRow> for DbLineItem {
+ fn from_row(row: &SqliteRow) -> sqlx::Result {
+ match row.try_get("item")? {
+ "checkin" => Ok(Self::Checkin(
+ row.try_get("item_date")?
+ )),
+ "invoke" => Ok(Self::CommandInvocation {
+ invoke_date: row.try_get("item_date")?,
+ user_name: row.try_get(2)?,
+ command_id: row.try_get(3)?,
+ action: row.try_get(4)?
+ }),
+ "result" => Ok(Self::CommandResult {
+ result_date: row.try_get("item_date")?,
+ user_name: row.try_get(5)?,
+ command_id: row.try_get(6)?,
+ action: row.try_get(7)?,
+ action_result: row.try_get(8)?
+ }),
+ type_name => Err(sqlx::Error::TypeNotFound {
+ type_name: type_name.to_string(),
+ }),
+ }
+ }
+ }
+
+ let info: Vec = sqlx::query_as(
+ r#"SELECT 'checkin' as item, checkin_date as item_date,
+ NULL, NULL, NULL,
+ NULL, NULL, NULL, NULL
+ FROM beacon_checkin WHERE beacon_id = ?
+ UNION
+ SELECT 'invoke' as item, bci.issue_date as item_date,
+ users.user_name, bci.command_id, cmd.cmd_parameters,
+ NULL, NULL, NULL, NULL
+ FROM beacon_command_invocation bci
+ INNER JOIN users ON users.user_id = bci.invoker_id
+ INNER JOIN beacon_command cmd ON cmd.command_id = bci.command_id
+ WHERE bci.beacon_id = ?
+ UNION
+ SELECT 'result' as item, bci.invocation_date as item_date,
+ NULL, NULL, NULL,
+ users.user_name, bci.command_id, cmd.cmd_parameters, bci.invocation_result
+ FROM beacon_command_invocation bci
+ INNER JOIN users ON users.user_id = bci.invoker_id
+ INNER JOIN beacon_command cmd ON cmd.command_id = bci.command_id
+ WHERE bci.beacon_id = ? AND bci.invocation_result IS NOT NULL
+ ORDER BY item_date DESC
+ LIMIT ?, 25"#,
+ )
+ .bind(&beacon_id)
+ .bind(&beacon_id)
+ .bind(&beacon_id)
+ .bind(offset)
+ .fetch_all(&state.db)
+ .await?;
+
+ let info = info
+ .into_iter()
+ .map(|r| match r {
+ DbLineItem::CommandInvocation {
+ user_name,
+ invoke_date,
+ command_id,
+ action
+ } => serde_json::from_str(&action)
+ .map_err(|_| InvokeOrResult::Invoke)
+ .map(|action| LineItem::CommandInvocation {
+ user_name,
+ invoke_date,
+ command_id,
+ action
+ }),
+ DbLineItem::CommandResult {
+ user_name,
+ result_date,
+ command_id,
+ action,
+ action_result
+ } => serde_json::from_str(&action)
+ .map_err(|_| InvokeOrResult::Result)
+ .map(|action| LineItem::CommandResult {
+ user_name,
+ result_date,
+ command_id,
+ action,
+ action_result
+ }),
+ DbLineItem::Checkin(d) => Ok(LineItem::Checkin(d)),
+ })
+ .collect();
+
+ send_event!(BeaconViewEvent::LoadHistorical(info));
+ }
+ Some(Ok(ws::Message::Close(_))) => {
+ break Ok(());
+ }
+ _ => {}
+ }
+ }
+ }
+}
pub async fn serve_web(
management_address: SocketAddrV4,
@@ -608,6 +834,10 @@ pub async fn serve_web(
"/api/subscribe/listener",
axum::routing::any(subscribe_to_listener_events)
)
+ .route(
+ "/api/subscribe/beacon/:beacon_id",
+ axum::routing::any(attach_to_beacon)
+ )
.route("/api/*fn_name", post(leptos_axum::handle_server_fns))
.leptos_routes_with_context(
&state,
diff --git a/sparse-server/style/beacons/_instances.scss b/sparse-server/style/beacons/_instances.scss
index de7ee8b..9925122 100644
--- a/sparse-server/style/beacons/_instances.scss
+++ b/sparse-server/style/beacons/_instances.scss
@@ -12,4 +12,59 @@ div.instance {
margin: 10px;
}
}
+
+ .instance-hist-beacon, .instance-hist-user {
+ margin: 15px 0px 0px 0px;
+ display: inline-block;
+ clear: both;
+ }
+
+ .instance-hist-bodiless {
+ padding: 15px;
+ }
+
+ .instance-hist-header {
+ padding: 15px;
+ margin: 0;
+ border-bottom: 1px solid;
+ }
+
+ .instance-hist-body {
+ padding: 15px;
+ margin: 0;
+ }
+
+ .instance-hist-beacon {
+ background-color: #11111c;
+ border: 1px solid #2e2e59;
+ float: left;
+
+ .instance-hist-header {
+ border-bottom-color: #2e2e59;
+ }
+ }
+
+ .instance-hist-user {
+ background-color: #2e2e59;
+ border: 1px solid #5e5ea0;
+ float: right;
+
+ .instance-hist-header {
+ border-bottom-color: #5e5e59;
+ }
+ }
+
+ .instance-hist-end {
+ padding: 15px;
+ clear: both;
+ margin: 15px;
+ display: flex;
+ justify-content: center;
+ width: 100%;
+ }
+
+ .instance-hist-end-inner {
+ font-style: italic;
+ display: inline-block;
+ }
}