Compare commits
2 Commits
6b70842b61
...
751e3b8654
| Author | SHA1 | Date | |
|---|---|---|---|
| 751e3b8654 | |||
| 24f06a78a9 |
@@ -1,7 +1,9 @@
|
||||
use crate::matrix_connection::sync_thread::MatrixSyncTaskID;
|
||||
use crate::users::{APIToken, UserEmail};
|
||||
use matrix_sdk::Room;
|
||||
use matrix_sdk::ruma::events::reaction::OriginalSyncReactionEvent;
|
||||
use matrix_sdk::ruma::events::room::message::OriginalSyncRoomMessageEvent;
|
||||
use matrix_sdk::ruma::events::room::redaction::OriginalSyncRoomRedactionEvent;
|
||||
use matrix_sdk::sync::SyncResponse;
|
||||
|
||||
pub type BroadcastSender = tokio::sync::broadcast::Sender<BroadcastMessage>;
|
||||
@@ -23,6 +25,18 @@ pub enum BroadcastMessage {
|
||||
event: Box<OriginalSyncRoomMessageEvent>,
|
||||
room: Room,
|
||||
},
|
||||
/// New reaction message
|
||||
ReactionEvent {
|
||||
user: UserEmail,
|
||||
event: Box<OriginalSyncReactionEvent>,
|
||||
room: Room,
|
||||
},
|
||||
/// New room redaction
|
||||
RoomRedactionEvent {
|
||||
user: UserEmail,
|
||||
event: Box<OriginalSyncRoomRedactionEvent>,
|
||||
room: Room,
|
||||
},
|
||||
/// Raw Matrix sync response
|
||||
MatrixSyncResponse { user: UserEmail, sync: SyncResponse },
|
||||
}
|
||||
|
||||
@@ -6,11 +6,13 @@ use crate::extractors::matrix_client_extractor::MatrixClientExtractor;
|
||||
use crate::matrix_connection::matrix_client::MatrixClient;
|
||||
use crate::matrix_connection::matrix_manager::MatrixManagerMsg;
|
||||
use actix_web::dev::Payload;
|
||||
use actix_web::{FromRequest, HttpRequest, web};
|
||||
use actix_web::{FromRequest, HttpRequest, HttpResponse, web};
|
||||
use actix_ws::Message;
|
||||
use futures_util::StreamExt;
|
||||
use matrix_sdk::ruma::OwnedRoomId;
|
||||
use matrix_sdk::ruma::events::reaction::ReactionEventContent;
|
||||
use matrix_sdk::ruma::events::room::message::RoomMessageEventContent;
|
||||
use matrix_sdk::ruma::events::room::redaction::RoomRedactionEventContent;
|
||||
use ractor::ActorRef;
|
||||
use std::time::Instant;
|
||||
use tokio::sync::broadcast;
|
||||
@@ -23,7 +25,19 @@ use tokio::time::interval;
|
||||
pub enum WsMessage {
|
||||
/// Room message event
|
||||
RoomMessageEvent {
|
||||
event: RoomMessageEventContent,
|
||||
event: Box<RoomMessageEventContent>,
|
||||
room_id: OwnedRoomId,
|
||||
},
|
||||
|
||||
/// Room reaction event
|
||||
RoomReactionEvent {
|
||||
event: Box<ReactionEventContent>,
|
||||
room_id: OwnedRoomId,
|
||||
},
|
||||
|
||||
/// Room reaction event
|
||||
RoomRedactionEvent {
|
||||
event: Box<RoomRedactionEventContent>,
|
||||
room_id: OwnedRoomId,
|
||||
},
|
||||
}
|
||||
@@ -38,6 +52,11 @@ pub async fn ws(
|
||||
// Forcefully ignore request payload by manually extracting authentication information
|
||||
let client = MatrixClientExtractor::from_request(&req, &mut Payload::None).await?;
|
||||
|
||||
// Check if Matrix link has been established first
|
||||
if !client.client.is_client_connected() {
|
||||
return Ok(HttpResponse::ExpectationFailed().json("Matrix link not established yet!"));
|
||||
}
|
||||
|
||||
// Ensure sync thread is started
|
||||
ractor::cast!(
|
||||
manager,
|
||||
@@ -114,12 +133,33 @@ pub async fn ws_handler(
|
||||
BroadcastMessage::RoomMessageEvent{user, event, room} if user == auth.user.email => {
|
||||
// Send the message to the websocket
|
||||
if let Ok(msg) = serde_json::to_string(&WsMessage::RoomMessageEvent {
|
||||
event:event.content,
|
||||
event: Box::new(event.content),
|
||||
room_id: room.room_id().to_owned(),
|
||||
}) && let Err(e) = session.text(msg).await {
|
||||
log::error!("Failed to send SyncEvent: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
BroadcastMessage::ReactionEvent{user, event, room} if user == auth.user.email => {
|
||||
// Send the message to the websocket
|
||||
if let Ok(msg) = serde_json::to_string(&WsMessage::RoomReactionEvent {
|
||||
event: Box::new(event.content),
|
||||
room_id: room.room_id().to_owned(),
|
||||
}) && let Err(e) = session.text(msg).await {
|
||||
log::error!("Failed to send SyncEvent: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
BroadcastMessage::RoomRedactionEvent{user, event, room} if user == auth.user.email => {
|
||||
// Send the message to the websocket
|
||||
if let Ok(msg) = serde_json::to_string(&WsMessage::RoomRedactionEvent {
|
||||
event: Box::new(event.content),
|
||||
room_id: room.room_id().to_owned(),
|
||||
}) && let Err(e) = session.text(msg).await {
|
||||
log::error!("Failed to send SyncEvent: {e}");
|
||||
}
|
||||
}
|
||||
|
||||
_ => {}
|
||||
};
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ impl MatrixClientExtractor {
|
||||
pub async fn to_extended_user_info(&self) -> anyhow::Result<ExtendedUserInfo> {
|
||||
Ok(ExtendedUserInfo {
|
||||
user: self.auth.user.clone(),
|
||||
matrix_account_connected: self.client.is_client_connected(),
|
||||
matrix_user_id: self.client.user_id().map(|id| id.to_string()),
|
||||
matrix_device_id: self.client.device_id().map(|id| id.to_string()),
|
||||
matrix_recovery_state: self.client.recovery_state(),
|
||||
|
||||
@@ -167,6 +167,9 @@ impl MatrixClient {
|
||||
.encryption()
|
||||
.wait_for_e2ee_initialization_tasks()
|
||||
.await;
|
||||
|
||||
// Save stored session once
|
||||
client.save_stored_session().await?;
|
||||
}
|
||||
|
||||
// Automatically save session when token gets refreshed
|
||||
|
||||
@@ -7,7 +7,9 @@ use crate::matrix_connection::matrix_client::MatrixClient;
|
||||
use crate::matrix_connection::matrix_manager::MatrixManagerMsg;
|
||||
use futures_util::StreamExt;
|
||||
use matrix_sdk::Room;
|
||||
use matrix_sdk::ruma::events::reaction::OriginalSyncReactionEvent;
|
||||
use matrix_sdk::ruma::events::room::message::OriginalSyncRoomMessageEvent;
|
||||
use matrix_sdk::ruma::events::room::redaction::OriginalSyncRoomRedactionEvent;
|
||||
use ractor::ActorRef;
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
@@ -48,19 +50,49 @@ async fn sync_thread_task(
|
||||
|
||||
let mut sync_stream = client.sync_stream().await;
|
||||
|
||||
let mut handlers = vec![];
|
||||
|
||||
let tx_msg_handle = tx.clone();
|
||||
let user = client.email.clone();
|
||||
let room_message_handle = client.add_event_handler(
|
||||
let user_msg_handle = client.email.clone();
|
||||
handlers.push(client.add_event_handler(
|
||||
async move |event: OriginalSyncRoomMessageEvent, room: Room| {
|
||||
if let Err(e) = tx_msg_handle.send(BroadcastMessage::RoomMessageEvent {
|
||||
user: user.clone(),
|
||||
user: user_msg_handle.clone(),
|
||||
event: Box::new(event),
|
||||
room,
|
||||
}) {
|
||||
log::warn!("Failed to forward room event! {e}");
|
||||
log::warn!("Failed to forward room message event! {e}");
|
||||
}
|
||||
},
|
||||
);
|
||||
));
|
||||
|
||||
let tx_reac_handle = tx.clone();
|
||||
let user_reac_handle = client.email.clone();
|
||||
handlers.push(client.add_event_handler(
|
||||
async move |event: OriginalSyncReactionEvent, room: Room| {
|
||||
if let Err(e) = tx_reac_handle.send(BroadcastMessage::ReactionEvent {
|
||||
user: user_reac_handle.clone(),
|
||||
event: Box::new(event),
|
||||
room,
|
||||
}) {
|
||||
log::warn!("Failed to forward reaction event! {e}");
|
||||
}
|
||||
},
|
||||
));
|
||||
|
||||
let tx_redac_handle = tx.clone();
|
||||
let user_redac_handle = client.email.clone();
|
||||
handlers.push(client.add_event_handler(
|
||||
async move |event: OriginalSyncRoomRedactionEvent, room: Room| {
|
||||
if let Err(e) = tx_redac_handle.send(BroadcastMessage::RoomRedactionEvent {
|
||||
user: user_redac_handle.clone(),
|
||||
event: Box::new(event),
|
||||
room,
|
||||
}) {
|
||||
log::warn!("Failed to forward reaction event! {e}");
|
||||
}
|
||||
},
|
||||
));
|
||||
|
||||
loop {
|
||||
tokio::select! {
|
||||
@@ -103,7 +135,9 @@ async fn sync_thread_task(
|
||||
}
|
||||
}
|
||||
|
||||
client.remove_event_handler(room_message_handle);
|
||||
for h in handlers {
|
||||
client.remove_event_handler(h);
|
||||
}
|
||||
|
||||
// Notify manager about termination, so this thread can be removed from the list
|
||||
log::info!("Sync thread {id:?} terminated!");
|
||||
|
||||
@@ -281,6 +281,7 @@ impl APIToken {
|
||||
pub struct ExtendedUserInfo {
|
||||
#[serde(flatten)]
|
||||
pub user: User,
|
||||
pub matrix_account_connected: bool,
|
||||
pub matrix_user_id: Option<String>,
|
||||
pub matrix_device_id: Option<String>,
|
||||
pub matrix_recovery_state: EncryptionRecoveryState,
|
||||
|
||||
@@ -6,6 +6,7 @@ export interface UserInfo {
|
||||
time_update: number;
|
||||
name: string;
|
||||
email: string;
|
||||
matrix_account_connected: boolean;
|
||||
matrix_user_id?: string;
|
||||
matrix_device_id?: string;
|
||||
matrix_recovery_state?: "Enabled" | "Disabled" | "Unknown" | "Incomplete";
|
||||
|
||||
@@ -6,7 +6,7 @@ import { NotLinkedAccountMessage } from "../widgets/NotLinkedAccountMessage";
|
||||
export function HomeRoute(): React.ReactElement {
|
||||
const user = useUserInfo();
|
||||
|
||||
if (!user.info.matrix_user_id) return <NotLinkedAccountMessage />;
|
||||
if (!user.info.matrix_account_connected) return <NotLinkedAccountMessage />;
|
||||
|
||||
return (
|
||||
<p>
|
||||
|
||||
@@ -4,7 +4,9 @@ import "react-json-view-lite/dist/index.css";
|
||||
import { WsApi, type WsMessage } from "../api/WsApi";
|
||||
import { useSnackbar } from "../hooks/contexts_provider/SnackbarProvider";
|
||||
import { time } from "../utils/DateUtils";
|
||||
import { useUserInfo } from "../widgets/dashboard/BaseAuthenticatedPage";
|
||||
import { MatrixGWRouteContainer } from "../widgets/MatrixGWRouteContainer";
|
||||
import { NotLinkedAccountMessage } from "../widgets/NotLinkedAccountMessage";
|
||||
|
||||
const State = {
|
||||
Closed: "Closed",
|
||||
@@ -15,6 +17,9 @@ const State = {
|
||||
type TimestampedMessages = WsMessage & { time: number };
|
||||
|
||||
export function WSDebugRoute(): React.ReactElement {
|
||||
const user = useUserInfo();
|
||||
if (!user.info.matrix_account_connected) return <NotLinkedAccountMessage />;
|
||||
|
||||
const snackbar = useSnackbar();
|
||||
|
||||
const [state, setState] = React.useState<string>(State.Closed);
|
||||
@@ -28,7 +33,7 @@ export function WSDebugRoute(): React.ReactElement {
|
||||
|
||||
ws.onopen = () => setState(State.Connected);
|
||||
ws.onerror = (e) => {
|
||||
console.error(`WS Debug error! ${e}`);
|
||||
console.error(`WS Debug error!`, e);
|
||||
snackbar(`WebSocket error! ${e}`);
|
||||
setState(State.Error);
|
||||
};
|
||||
|
||||
@@ -8,6 +8,7 @@ import { useTheme } from "@mui/material/styles";
|
||||
import type {} from "@mui/material/themeCssVarsAugmentation";
|
||||
import useMediaQuery from "@mui/material/useMediaQuery";
|
||||
import * as React from "react";
|
||||
import { useUserInfo } from "./BaseAuthenticatedPage";
|
||||
import DashboardSidebarContext from "./DashboardSidebarContext";
|
||||
import DashboardSidebarDividerItem from "./DashboardSidebarDividerItem";
|
||||
import DashboardSidebarPageItem from "./DashboardSidebarPageItem";
|
||||
@@ -31,6 +32,7 @@ export default function DashboardSidebar({
|
||||
container,
|
||||
}: DashboardSidebarProps) {
|
||||
const theme = useTheme();
|
||||
const user = useUserInfo();
|
||||
|
||||
const isOverSmViewport = useMediaQuery(theme.breakpoints.up("sm"));
|
||||
const isOverMdViewport = useMediaQuery(theme.breakpoints.up("md"));
|
||||
@@ -99,6 +101,7 @@ export default function DashboardSidebar({
|
||||
}}
|
||||
>
|
||||
<DashboardSidebarPageItem
|
||||
disabled={!user.info.matrix_account_connected}
|
||||
title="Messages"
|
||||
icon={<Icon path={mdiForum} size={"1.5em"} />}
|
||||
href="/"
|
||||
@@ -115,6 +118,7 @@ export default function DashboardSidebar({
|
||||
href="/tokens"
|
||||
/>
|
||||
<DashboardSidebarPageItem
|
||||
disabled={!user.info.matrix_account_connected}
|
||||
title="WS Debug"
|
||||
icon={<Icon path={mdiBug} size={"1.5em"} />}
|
||||
href="/wsdebug"
|
||||
@@ -123,7 +127,12 @@ export default function DashboardSidebar({
|
||||
</Box>
|
||||
</React.Fragment>
|
||||
),
|
||||
[mini, hasDrawerTransitions, isFullyExpanded]
|
||||
[
|
||||
mini,
|
||||
hasDrawerTransitions,
|
||||
isFullyExpanded,
|
||||
user.info.matrix_account_connected,
|
||||
]
|
||||
);
|
||||
|
||||
const getDrawerSharedSx = React.useCallback(
|
||||
|
||||
Reference in New Issue
Block a user