1 use actix_web::{web, Error, HttpRequest, HttpResponse};
2 use actix_web_actors::ws;
3 use actix_ws::{MessageStream, Session};
4 use futures::stream::StreamExt;
5 use lemmy_api::Perform;
6 use lemmy_api_common::{
34 context::LemmyContext,
50 MarkCommentReplyAsRead,
51 MarkPersonMentionAsRead,
52 PasswordChangeAfterReset,
77 CreatePrivateMessageReport,
81 ListPrivateMessageReports,
82 MarkPrivateMessageAsRead,
83 ResolvePrivateMessageReport,
86 ApproveRegistrationApplication,
91 GetUnreadRegistrationApplicationCount,
93 ListRegistrationApplications,
102 serialize_websocket_message,
103 structs::{CommunityJoin, ModJoin, PostJoin, UserJoin},
109 use lemmy_api_crud::PerformCrud;
110 use lemmy_apub::{api::PerformApub, SendActivity};
111 use lemmy_utils::{error::LemmyError, rate_limit::RateLimitCell, ConnectionId, IpAddr};
112 use serde::Deserialize;
113 use serde_json::Value;
118 time::{Duration, Instant},
120 use tracing::{debug, error, info};
122 /// Entry point for our route
123 pub async fn websocket(
126 context: web::Data<LemmyContext>,
127 rate_limiter: web::Data<RateLimitCell>,
128 ) -> Result<HttpResponse, Error> {
129 let (response, session, stream) = actix_ws::handle(&req, body)?;
131 let client_ip = IpAddr(
134 .realip_remote_addr()
135 .unwrap_or("blank_ip")
139 let check = rate_limiter.message().check(client_ip.clone());
142 "Websocket join with IP: {} has been rate limited.",
145 session.close(None).await.map_err(LemmyError::from)?;
149 let connection_id = context.chat_server().handle_connect(session.clone())?;
150 info!("{} joined", &client_ip);
152 let alive = Arc::new(Mutex::new(Instant::now()));
153 heartbeat(session.clone(), alive.clone());
155 actix_rt::spawn(handle_messages(
168 async fn handle_messages(
169 mut stream: MessageStream,
171 mut session: Session,
172 connection_id: ConnectionId,
173 alive: Arc<Mutex<Instant>>,
174 rate_limiter: web::Data<RateLimitCell>,
175 context: web::Data<LemmyContext>,
176 ) -> Result<(), LemmyError> {
177 while let Some(Ok(msg)) = stream.next().await {
179 ws::Message::Ping(bytes) => {
180 if session.pong(&bytes).await.is_err() {
184 ws::Message::Pong(_) => {
187 .expect("Failed to acquire websocket heartbeat alive lock");
188 *lock = Instant::now();
190 ws::Message::Text(text) => {
191 let msg = text.trim().to_string();
192 let executed = parse_json_message(
196 rate_limiter.get_ref(),
197 context.get_ref().clone(),
201 let res = executed.unwrap_or_else(|e| {
202 error!("Error during message handling {}", e);
204 .unwrap_or_else(|_| String::from(r#"{"error":"failed to serialize json"}"#))
206 session.text(res).await?;
208 ws::Message::Close(_) => {
209 session.close(None).await?;
210 context.chat_server().handle_disconnect(&connection_id)?;
213 ws::Message::Binary(_) => info!("Unexpected binary"),
220 fn heartbeat(mut session: Session, alive: Arc<Mutex<Instant>>) {
221 actix_rt::spawn(async move {
222 let mut interval = actix_rt::time::interval(Duration::from_secs(5));
224 if session.ping(b"").await.is_err() {
228 let duration_since = {
229 let alive_lock = alive
231 .expect("Failed to acquire websocket heartbeat alive lock");
232 Instant::now().duration_since(*alive_lock)
234 if duration_since > Duration::from_secs(10) {
235 let _ = session.close(None).await;
238 interval.tick().await;
243 async fn parse_json_message(
246 connection_id: ConnectionId,
247 rate_limiter: &RateLimitCell,
248 context: LemmyContext,
249 ) -> Result<String, LemmyError> {
250 let json: Value = serde_json::from_str(&msg)?;
251 let data = &json["data"].to_string();
254 .ok_or_else(|| LemmyError::from_message("missing op"))?;
256 // check if api call passes the rate limit, and generate future for later execution
257 if let Ok(user_operation_crud) = UserOperationCrud::from_str(op) {
258 let passed = match user_operation_crud {
259 UserOperationCrud::Register => rate_limiter.register().check(ip),
260 UserOperationCrud::CreatePost => rate_limiter.post().check(ip),
261 UserOperationCrud::CreateCommunity => rate_limiter.register().check(ip),
262 UserOperationCrud::CreateComment => rate_limiter.comment().check(ip),
263 _ => rate_limiter.message().check(ip),
265 check_rate_limit_passed(passed)?;
266 match_websocket_operation_crud(context, connection_id, user_operation_crud, data).await
267 } else if let Ok(user_operation) = UserOperation::from_str(op) {
268 let passed = match user_operation {
269 UserOperation::GetCaptcha => rate_limiter.post().check(ip),
270 _ => rate_limiter.message().check(ip),
272 check_rate_limit_passed(passed)?;
273 match_websocket_operation(context, connection_id, user_operation, data).await
275 let user_operation = UserOperationApub::from_str(op)?;
276 let passed = match user_operation {
277 UserOperationApub::Search => rate_limiter.search().check(ip),
278 _ => rate_limiter.message().check(ip),
280 check_rate_limit_passed(passed)?;
281 match_websocket_operation_apub(context, connection_id, user_operation, data).await
285 fn check_rate_limit_passed(passed: bool) -> Result<(), LemmyError> {
289 // if rate limit was hit, respond with message
290 Err(LemmyError::from_message("rate_limit_error"))
294 pub async fn match_websocket_operation_crud(
295 context: LemmyContext,
297 op: UserOperationCrud,
299 ) -> result::Result<String, LemmyError> {
302 UserOperationCrud::Register => {
303 do_websocket_operation_crud::<Register>(context, id, op, data).await
305 UserOperationCrud::DeleteAccount => {
306 do_websocket_operation_crud::<DeleteAccount>(context, id, op, data).await
309 // Private Message ops
310 UserOperationCrud::CreatePrivateMessage => {
311 do_websocket_operation_crud::<CreatePrivateMessage>(context, id, op, data).await
313 UserOperationCrud::EditPrivateMessage => {
314 do_websocket_operation_crud::<EditPrivateMessage>(context, id, op, data).await
316 UserOperationCrud::DeletePrivateMessage => {
317 do_websocket_operation_crud::<DeletePrivateMessage>(context, id, op, data).await
319 UserOperationCrud::GetPrivateMessages => {
320 do_websocket_operation_crud::<GetPrivateMessages>(context, id, op, data).await
324 UserOperationCrud::CreateSite => {
325 do_websocket_operation_crud::<CreateSite>(context, id, op, data).await
327 UserOperationCrud::EditSite => {
328 do_websocket_operation_crud::<EditSite>(context, id, op, data).await
330 UserOperationCrud::GetSite => {
331 do_websocket_operation_crud::<GetSite>(context, id, op, data).await
335 UserOperationCrud::ListCommunities => {
336 do_websocket_operation_crud::<ListCommunities>(context, id, op, data).await
338 UserOperationCrud::CreateCommunity => {
339 do_websocket_operation_crud::<CreateCommunity>(context, id, op, data).await
341 UserOperationCrud::EditCommunity => {
342 do_websocket_operation_crud::<EditCommunity>(context, id, op, data).await
344 UserOperationCrud::DeleteCommunity => {
345 do_websocket_operation_crud::<DeleteCommunity>(context, id, op, data).await
347 UserOperationCrud::RemoveCommunity => {
348 do_websocket_operation_crud::<RemoveCommunity>(context, id, op, data).await
352 UserOperationCrud::CreatePost => {
353 do_websocket_operation_crud::<CreatePost>(context, id, op, data).await
355 UserOperationCrud::GetPost => {
356 do_websocket_operation_crud::<GetPost>(context, id, op, data).await
358 UserOperationCrud::EditPost => {
359 do_websocket_operation_crud::<EditPost>(context, id, op, data).await
361 UserOperationCrud::DeletePost => {
362 do_websocket_operation_crud::<DeletePost>(context, id, op, data).await
364 UserOperationCrud::RemovePost => {
365 do_websocket_operation_crud::<RemovePost>(context, id, op, data).await
369 UserOperationCrud::CreateComment => {
370 do_websocket_operation_crud::<CreateComment>(context, id, op, data).await
372 UserOperationCrud::EditComment => {
373 do_websocket_operation_crud::<EditComment>(context, id, op, data).await
375 UserOperationCrud::DeleteComment => {
376 do_websocket_operation_crud::<DeleteComment>(context, id, op, data).await
378 UserOperationCrud::RemoveComment => {
379 do_websocket_operation_crud::<RemoveComment>(context, id, op, data).await
381 UserOperationCrud::GetComment => {
382 do_websocket_operation_crud::<GetComment>(context, id, op, data).await
387 async fn do_websocket_operation_crud<'a, 'b, Data>(
388 context: LemmyContext,
390 op: UserOperationCrud,
392 ) -> result::Result<String, LemmyError>
394 Data: PerformCrud + SendActivity<Response = <Data as PerformCrud>::Response>,
395 for<'de> Data: Deserialize<'de>,
397 let parsed_data: Data = serde_json::from_str(data)?;
398 let res = parsed_data
399 .perform(&web::Data::new(context.clone()), Some(id))
401 SendActivity::send_activity(&parsed_data, &res, &context).await?;
402 serialize_websocket_message(&op, &res)
405 pub async fn match_websocket_operation_apub(
406 context: LemmyContext,
408 op: UserOperationApub,
410 ) -> result::Result<String, LemmyError> {
412 UserOperationApub::GetPersonDetails => {
413 do_websocket_operation_apub::<GetPersonDetails>(context, id, op, data).await
415 UserOperationApub::GetCommunity => {
416 do_websocket_operation_apub::<GetCommunity>(context, id, op, data).await
418 UserOperationApub::GetComments => {
419 do_websocket_operation_apub::<GetComments>(context, id, op, data).await
421 UserOperationApub::GetPosts => {
422 do_websocket_operation_apub::<GetPosts>(context, id, op, data).await
424 UserOperationApub::ResolveObject => {
425 do_websocket_operation_apub::<ResolveObject>(context, id, op, data).await
427 UserOperationApub::Search => do_websocket_operation_apub::<Search>(context, id, op, data).await,
431 async fn do_websocket_operation_apub<'a, 'b, Data>(
432 context: LemmyContext,
434 op: UserOperationApub,
436 ) -> result::Result<String, LemmyError>
438 Data: PerformApub + SendActivity<Response = <Data as PerformApub>::Response>,
439 for<'de> Data: Deserialize<'de>,
441 let parsed_data: Data = serde_json::from_str(data)?;
442 let res = parsed_data
443 .perform(&web::Data::new(context.clone()), Some(id))
445 SendActivity::send_activity(&parsed_data, &res, &context).await?;
446 serialize_websocket_message(&op, &res)
449 pub async fn match_websocket_operation(
450 context: LemmyContext,
454 ) -> result::Result<String, LemmyError> {
457 UserOperation::Login => do_websocket_operation::<Login>(context, id, op, data).await,
458 UserOperation::GetCaptcha => do_websocket_operation::<GetCaptcha>(context, id, op, data).await,
459 UserOperation::GetReplies => do_websocket_operation::<GetReplies>(context, id, op, data).await,
460 UserOperation::AddAdmin => do_websocket_operation::<AddAdmin>(context, id, op, data).await,
461 UserOperation::GetUnreadRegistrationApplicationCount => {
462 do_websocket_operation::<GetUnreadRegistrationApplicationCount>(context, id, op, data).await
464 UserOperation::ListRegistrationApplications => {
465 do_websocket_operation::<ListRegistrationApplications>(context, id, op, data).await
467 UserOperation::ApproveRegistrationApplication => {
468 do_websocket_operation::<ApproveRegistrationApplication>(context, id, op, data).await
470 UserOperation::BanPerson => do_websocket_operation::<BanPerson>(context, id, op, data).await,
471 UserOperation::GetBannedPersons => {
472 do_websocket_operation::<GetBannedPersons>(context, id, op, data).await
474 UserOperation::BlockPerson => {
475 do_websocket_operation::<BlockPerson>(context, id, op, data).await
477 UserOperation::GetPersonMentions => {
478 do_websocket_operation::<GetPersonMentions>(context, id, op, data).await
480 UserOperation::MarkPersonMentionAsRead => {
481 do_websocket_operation::<MarkPersonMentionAsRead>(context, id, op, data).await
483 UserOperation::MarkCommentReplyAsRead => {
484 do_websocket_operation::<MarkCommentReplyAsRead>(context, id, op, data).await
486 UserOperation::MarkAllAsRead => {
487 do_websocket_operation::<MarkAllAsRead>(context, id, op, data).await
489 UserOperation::PasswordReset => {
490 do_websocket_operation::<PasswordReset>(context, id, op, data).await
492 UserOperation::PasswordChange => {
493 do_websocket_operation::<PasswordChangeAfterReset>(context, id, op, data).await
495 UserOperation::UserJoin => do_websocket_operation::<UserJoin>(context, id, op, data).await,
496 UserOperation::PostJoin => do_websocket_operation::<PostJoin>(context, id, op, data).await,
497 UserOperation::CommunityJoin => {
498 do_websocket_operation::<CommunityJoin>(context, id, op, data).await
500 UserOperation::ModJoin => do_websocket_operation::<ModJoin>(context, id, op, data).await,
501 UserOperation::SaveUserSettings => {
502 do_websocket_operation::<SaveUserSettings>(context, id, op, data).await
504 UserOperation::ChangePassword => {
505 do_websocket_operation::<ChangePassword>(context, id, op, data).await
507 UserOperation::GetReportCount => {
508 do_websocket_operation::<GetReportCount>(context, id, op, data).await
510 UserOperation::GetUnreadCount => {
511 do_websocket_operation::<GetUnreadCount>(context, id, op, data).await
513 UserOperation::VerifyEmail => {
514 do_websocket_operation::<VerifyEmail>(context, id, op, data).await
517 // Private Message ops
518 UserOperation::MarkPrivateMessageAsRead => {
519 do_websocket_operation::<MarkPrivateMessageAsRead>(context, id, op, data).await
521 UserOperation::CreatePrivateMessageReport => {
522 do_websocket_operation::<CreatePrivateMessageReport>(context, id, op, data).await
524 UserOperation::ResolvePrivateMessageReport => {
525 do_websocket_operation::<ResolvePrivateMessageReport>(context, id, op, data).await
527 UserOperation::ListPrivateMessageReports => {
528 do_websocket_operation::<ListPrivateMessageReports>(context, id, op, data).await
532 UserOperation::GetModlog => do_websocket_operation::<GetModlog>(context, id, op, data).await,
533 UserOperation::PurgePerson => {
534 do_websocket_operation::<PurgePerson>(context, id, op, data).await
536 UserOperation::PurgeCommunity => {
537 do_websocket_operation::<PurgeCommunity>(context, id, op, data).await
539 UserOperation::PurgePost => do_websocket_operation::<PurgePost>(context, id, op, data).await,
540 UserOperation::PurgeComment => {
541 do_websocket_operation::<PurgeComment>(context, id, op, data).await
543 UserOperation::TransferCommunity => {
544 do_websocket_operation::<TransferCommunity>(context, id, op, data).await
546 UserOperation::LeaveAdmin => do_websocket_operation::<LeaveAdmin>(context, id, op, data).await,
549 UserOperation::FollowCommunity => {
550 do_websocket_operation::<FollowCommunity>(context, id, op, data).await
552 UserOperation::BlockCommunity => {
553 do_websocket_operation::<BlockCommunity>(context, id, op, data).await
555 UserOperation::BanFromCommunity => {
556 do_websocket_operation::<BanFromCommunity>(context, id, op, data).await
558 UserOperation::AddModToCommunity => {
559 do_websocket_operation::<AddModToCommunity>(context, id, op, data).await
563 UserOperation::LockPost => do_websocket_operation::<LockPost>(context, id, op, data).await,
564 UserOperation::FeaturePost => {
565 do_websocket_operation::<FeaturePost>(context, id, op, data).await
567 UserOperation::CreatePostLike => {
568 do_websocket_operation::<CreatePostLike>(context, id, op, data).await
570 UserOperation::MarkPostAsRead => {
571 do_websocket_operation::<MarkPostAsRead>(context, id, op, data).await
573 UserOperation::SavePost => do_websocket_operation::<SavePost>(context, id, op, data).await,
574 UserOperation::CreatePostReport => {
575 do_websocket_operation::<CreatePostReport>(context, id, op, data).await
577 UserOperation::ListPostReports => {
578 do_websocket_operation::<ListPostReports>(context, id, op, data).await
580 UserOperation::ResolvePostReport => {
581 do_websocket_operation::<ResolvePostReport>(context, id, op, data).await
583 UserOperation::GetSiteMetadata => {
584 do_websocket_operation::<GetSiteMetadata>(context, id, op, data).await
588 UserOperation::SaveComment => {
589 do_websocket_operation::<SaveComment>(context, id, op, data).await
591 UserOperation::CreateCommentLike => {
592 do_websocket_operation::<CreateCommentLike>(context, id, op, data).await
594 UserOperation::DistinguishComment => {
595 do_websocket_operation::<DistinguishComment>(context, id, op, data).await
597 UserOperation::CreateCommentReport => {
598 do_websocket_operation::<CreateCommentReport>(context, id, op, data).await
600 UserOperation::ListCommentReports => {
601 do_websocket_operation::<ListCommentReports>(context, id, op, data).await
603 UserOperation::ResolveCommentReport => {
604 do_websocket_operation::<ResolveCommentReport>(context, id, op, data).await
609 async fn do_websocket_operation<'a, 'b, Data>(
610 context: LemmyContext,
614 ) -> result::Result<String, LemmyError>
616 Data: Perform + SendActivity<Response = <Data as Perform>::Response>,
617 for<'de> Data: Deserialize<'de>,
619 let parsed_data: Data = serde_json::from_str(data)?;
620 let res = parsed_data
621 .perform(&web::Data::new(context.clone()), Some(id))
623 SendActivity::send_activity(&parsed_data, &res, &context).await?;
624 serialize_websocket_message(&op, &res)