1 use activitypub_federation::config::Data as ContextData;
2 use actix_web::{web, Error, HttpRequest, HttpResponse};
3 use actix_web_actors::ws;
4 use actix_ws::{MessageStream, Session};
5 use futures::stream::StreamExt;
6 use lemmy_api::Perform;
7 use lemmy_api_common::{
35 context::LemmyContext,
36 custom_emoji::{CreateCustomEmoji, DeleteCustomEmoji, EditCustomEmoji},
52 MarkCommentReplyAsRead,
53 MarkPersonMentionAsRead,
54 PasswordChangeAfterReset,
79 CreatePrivateMessageReport,
83 ListPrivateMessageReports,
84 MarkPrivateMessageAsRead,
85 ResolvePrivateMessageReport,
88 ApproveRegistrationApplication,
93 GetUnreadRegistrationApplicationCount,
95 ListRegistrationApplications,
104 serialize_websocket_message,
105 structs::{CommunityJoin, ModJoin, PostJoin, UserJoin},
111 use lemmy_api_crud::PerformCrud;
112 use lemmy_apub::{api::PerformApub, SendActivity};
113 use lemmy_utils::{error::LemmyError, rate_limit::RateLimitCell, ConnectionId, IpAddr};
114 use serde::Deserialize;
115 use serde_json::Value;
121 time::{Duration, Instant},
123 use tracing::{debug, error, info};
125 /// Entry point for our route
126 pub async fn websocket(
129 context: web::Data<LemmyContext>,
130 rate_limiter: web::Data<RateLimitCell>,
131 apub_data: ContextData<LemmyContext>,
132 ) -> Result<HttpResponse, Error> {
133 let (response, session, stream) = actix_ws::handle(&req, body)?;
135 let client_ip = IpAddr(
138 .realip_remote_addr()
139 .unwrap_or("blank_ip")
143 let check = rate_limiter.message().check(client_ip.clone());
146 "Websocket join with IP: {} has been rate limited.",
149 session.close(None).await.map_err(LemmyError::from)?;
153 let connection_id = context.chat_server().handle_connect(session.clone())?;
154 info!("{} joined", &client_ip);
156 let alive = Arc::new(Mutex::new(Instant::now()));
157 heartbeat(session.clone(), alive.clone());
159 actix_rt::spawn(handle_messages(
172 async fn handle_messages(
173 mut stream: MessageStream,
175 mut session: Session,
176 connection_id: ConnectionId,
177 alive: Arc<Mutex<Instant>>,
178 rate_limiter: web::Data<RateLimitCell>,
179 context: ContextData<LemmyContext>,
180 ) -> Result<(), LemmyError> {
181 while let Some(Ok(msg)) = stream.next().await {
183 ws::Message::Ping(bytes) => {
184 if session.pong(&bytes).await.is_err() {
188 ws::Message::Pong(_) => {
191 .expect("Failed to acquire websocket heartbeat alive lock");
192 *lock = Instant::now();
194 ws::Message::Text(text) => {
195 let msg = text.trim().to_string();
196 let executed = parse_json_message(
200 rate_limiter.get_ref(),
201 context.reset_request_count(),
205 let res = executed.unwrap_or_else(|e| {
206 error!("Error during message handling {}", e);
208 .unwrap_or_else(|_| String::from(r#"{"error":"failed to serialize json"}"#))
210 session.text(res).await?;
212 ws::Message::Close(_) => {
213 session.close(None).await?;
214 context.chat_server().handle_disconnect(&connection_id)?;
217 ws::Message::Binary(_) => info!("Unexpected binary"),
224 fn heartbeat(mut session: Session, alive: Arc<Mutex<Instant>>) {
225 actix_rt::spawn(async move {
226 let mut interval = actix_rt::time::interval(Duration::from_secs(5));
228 if session.ping(b"").await.is_err() {
232 let duration_since = {
233 let alive_lock = alive
235 .expect("Failed to acquire websocket heartbeat alive lock");
236 Instant::now().duration_since(*alive_lock)
238 if duration_since > Duration::from_secs(10) {
239 let _ = session.close(None).await;
242 interval.tick().await;
247 async fn parse_json_message(
250 connection_id: ConnectionId,
251 rate_limiter: &RateLimitCell,
252 context: ContextData<LemmyContext>,
253 ) -> Result<String, LemmyError> {
254 let json: Value = serde_json::from_str(&msg)?;
258 .ok_or_else(|| LemmyError::from_message("missing data"))?;
260 let missing_op_err = || LemmyError::from_message("missing op");
264 .ok_or_else(missing_op_err)?
266 .ok_or_else(missing_op_err)?;
268 // check if api call passes the rate limit, and generate future for later execution
269 if let Ok(user_operation_crud) = UserOperationCrud::from_str(op) {
270 let passed = match user_operation_crud {
271 UserOperationCrud::Register => rate_limiter.register().check(ip),
272 UserOperationCrud::CreatePost => rate_limiter.post().check(ip),
273 UserOperationCrud::CreateCommunity => rate_limiter.register().check(ip),
274 UserOperationCrud::CreateComment => rate_limiter.comment().check(ip),
275 _ => rate_limiter.message().check(ip),
277 check_rate_limit_passed(passed)?;
278 match_websocket_operation_crud(context, connection_id, user_operation_crud, data).await
279 } else if let Ok(user_operation) = UserOperation::from_str(op) {
280 let passed = match user_operation {
281 UserOperation::GetCaptcha => rate_limiter.post().check(ip),
282 _ => rate_limiter.message().check(ip),
284 check_rate_limit_passed(passed)?;
285 match_websocket_operation(context, connection_id, user_operation, data).await
287 let user_operation = UserOperationApub::from_str(op)?;
288 let passed = match user_operation {
289 UserOperationApub::Search => rate_limiter.search().check(ip),
290 _ => rate_limiter.message().check(ip),
292 check_rate_limit_passed(passed)?;
293 match_websocket_operation_apub(context, connection_id, user_operation, data).await
297 fn check_rate_limit_passed(passed: bool) -> Result<(), LemmyError> {
301 // if rate limit was hit, respond with message
302 Err(LemmyError::from_message("rate_limit_error"))
306 pub async fn match_websocket_operation_crud(
307 context: ContextData<LemmyContext>,
309 op: UserOperationCrud,
311 ) -> result::Result<String, LemmyError> {
314 UserOperationCrud::Register => {
315 do_websocket_operation_crud::<Register>(context, id, op, data).await
317 UserOperationCrud::DeleteAccount => {
318 do_websocket_operation_crud::<DeleteAccount>(context, id, op, data).await
321 // Private Message ops
322 UserOperationCrud::CreatePrivateMessage => {
323 do_websocket_operation_crud::<CreatePrivateMessage>(context, id, op, data).await
325 UserOperationCrud::EditPrivateMessage => {
326 do_websocket_operation_crud::<EditPrivateMessage>(context, id, op, data).await
328 UserOperationCrud::DeletePrivateMessage => {
329 do_websocket_operation_crud::<DeletePrivateMessage>(context, id, op, data).await
331 UserOperationCrud::GetPrivateMessages => {
332 do_websocket_operation_crud::<GetPrivateMessages>(context, id, op, data).await
336 UserOperationCrud::CreateSite => {
337 do_websocket_operation_crud::<CreateSite>(context, id, op, data).await
339 UserOperationCrud::EditSite => {
340 do_websocket_operation_crud::<EditSite>(context, id, op, data).await
342 UserOperationCrud::GetSite => {
343 do_websocket_operation_crud::<GetSite>(context, id, op, data).await
347 UserOperationCrud::ListCommunities => {
348 do_websocket_operation_crud::<ListCommunities>(context, id, op, data).await
350 UserOperationCrud::CreateCommunity => {
351 do_websocket_operation_crud::<CreateCommunity>(context, id, op, data).await
353 UserOperationCrud::EditCommunity => {
354 do_websocket_operation_crud::<EditCommunity>(context, id, op, data).await
356 UserOperationCrud::DeleteCommunity => {
357 do_websocket_operation_crud::<DeleteCommunity>(context, id, op, data).await
359 UserOperationCrud::RemoveCommunity => {
360 do_websocket_operation_crud::<RemoveCommunity>(context, id, op, data).await
364 UserOperationCrud::CreatePost => {
365 do_websocket_operation_crud::<CreatePost>(context, id, op, data).await
367 UserOperationCrud::GetPost => {
368 do_websocket_operation_crud::<GetPost>(context, id, op, data).await
370 UserOperationCrud::EditPost => {
371 do_websocket_operation_crud::<EditPost>(context, id, op, data).await
373 UserOperationCrud::DeletePost => {
374 do_websocket_operation_crud::<DeletePost>(context, id, op, data).await
376 UserOperationCrud::RemovePost => {
377 do_websocket_operation_crud::<RemovePost>(context, id, op, data).await
381 UserOperationCrud::CreateComment => {
382 do_websocket_operation_crud::<CreateComment>(context, id, op, data).await
384 UserOperationCrud::EditComment => {
385 do_websocket_operation_crud::<EditComment>(context, id, op, data).await
387 UserOperationCrud::DeleteComment => {
388 do_websocket_operation_crud::<DeleteComment>(context, id, op, data).await
390 UserOperationCrud::RemoveComment => {
391 do_websocket_operation_crud::<RemoveComment>(context, id, op, data).await
393 UserOperationCrud::GetComment => {
394 do_websocket_operation_crud::<GetComment>(context, id, op, data).await
397 UserOperationCrud::CreateCustomEmoji => {
398 do_websocket_operation_crud::<CreateCustomEmoji>(context, id, op, data).await
400 UserOperationCrud::EditCustomEmoji => {
401 do_websocket_operation_crud::<EditCustomEmoji>(context, id, op, data).await
403 UserOperationCrud::DeleteCustomEmoji => {
404 do_websocket_operation_crud::<DeleteCustomEmoji>(context, id, op, data).await
409 async fn do_websocket_operation_crud<'a, 'b, Data>(
410 context: ContextData<LemmyContext>,
412 op: UserOperationCrud,
414 ) -> result::Result<String, LemmyError>
416 Data: PerformCrud + SendActivity<Response = <Data as PerformCrud>::Response> + Send,
417 for<'de> Data: Deserialize<'de>,
419 let parsed_data: Data = serde_json::from_value(data)?;
420 let res = parsed_data
421 .perform(&web::Data::new(context.deref().clone()), Some(id))
423 SendActivity::send_activity(&parsed_data, &res, &context).await?;
424 serialize_websocket_message(&op, &res)
427 pub async fn match_websocket_operation_apub(
428 context: ContextData<LemmyContext>,
430 op: UserOperationApub,
432 ) -> result::Result<String, LemmyError> {
434 UserOperationApub::GetPersonDetails => {
435 do_websocket_operation_apub::<GetPersonDetails>(context, id, op, data).await
437 UserOperationApub::GetCommunity => {
438 do_websocket_operation_apub::<GetCommunity>(context, id, op, data).await
440 UserOperationApub::GetComments => {
441 do_websocket_operation_apub::<GetComments>(context, id, op, data).await
443 UserOperationApub::GetPosts => {
444 do_websocket_operation_apub::<GetPosts>(context, id, op, data).await
446 UserOperationApub::ResolveObject => {
447 do_websocket_operation_apub::<ResolveObject>(context, id, op, data).await
449 UserOperationApub::Search => do_websocket_operation_apub::<Search>(context, id, op, data).await,
453 async fn do_websocket_operation_apub<'a, 'b, Data>(
454 context: ContextData<LemmyContext>,
456 op: UserOperationApub,
458 ) -> result::Result<String, LemmyError>
460 Data: PerformApub + SendActivity<Response = <Data as PerformApub>::Response> + Send,
461 for<'de> Data: Deserialize<'de>,
463 let parsed_data: Data = serde_json::from_value(data)?;
464 let res = parsed_data.perform(&context, Some(id)).await?;
465 SendActivity::send_activity(&parsed_data, &res, &context).await?;
466 serialize_websocket_message(&op, &res)
469 pub async fn match_websocket_operation(
470 context: ContextData<LemmyContext>,
474 ) -> result::Result<String, LemmyError> {
477 UserOperation::Login => do_websocket_operation::<Login>(context, id, op, data).await,
478 UserOperation::GetCaptcha => do_websocket_operation::<GetCaptcha>(context, id, op, data).await,
479 UserOperation::GetReplies => do_websocket_operation::<GetReplies>(context, id, op, data).await,
480 UserOperation::AddAdmin => do_websocket_operation::<AddAdmin>(context, id, op, data).await,
481 UserOperation::GetUnreadRegistrationApplicationCount => {
482 do_websocket_operation::<GetUnreadRegistrationApplicationCount>(context, id, op, data).await
484 UserOperation::ListRegistrationApplications => {
485 do_websocket_operation::<ListRegistrationApplications>(context, id, op, data).await
487 UserOperation::ApproveRegistrationApplication => {
488 do_websocket_operation::<ApproveRegistrationApplication>(context, id, op, data).await
490 UserOperation::BanPerson => do_websocket_operation::<BanPerson>(context, id, op, data).await,
491 UserOperation::GetBannedPersons => {
492 do_websocket_operation::<GetBannedPersons>(context, id, op, data).await
494 UserOperation::BlockPerson => {
495 do_websocket_operation::<BlockPerson>(context, id, op, data).await
497 UserOperation::GetPersonMentions => {
498 do_websocket_operation::<GetPersonMentions>(context, id, op, data).await
500 UserOperation::MarkPersonMentionAsRead => {
501 do_websocket_operation::<MarkPersonMentionAsRead>(context, id, op, data).await
503 UserOperation::MarkCommentReplyAsRead => {
504 do_websocket_operation::<MarkCommentReplyAsRead>(context, id, op, data).await
506 UserOperation::MarkAllAsRead => {
507 do_websocket_operation::<MarkAllAsRead>(context, id, op, data).await
509 UserOperation::PasswordReset => {
510 do_websocket_operation::<PasswordReset>(context, id, op, data).await
512 UserOperation::PasswordChange => {
513 do_websocket_operation::<PasswordChangeAfterReset>(context, id, op, data).await
515 UserOperation::UserJoin => do_websocket_operation::<UserJoin>(context, id, op, data).await,
516 UserOperation::PostJoin => do_websocket_operation::<PostJoin>(context, id, op, data).await,
517 UserOperation::CommunityJoin => {
518 do_websocket_operation::<CommunityJoin>(context, id, op, data).await
520 UserOperation::ModJoin => do_websocket_operation::<ModJoin>(context, id, op, data).await,
521 UserOperation::SaveUserSettings => {
522 do_websocket_operation::<SaveUserSettings>(context, id, op, data).await
524 UserOperation::ChangePassword => {
525 do_websocket_operation::<ChangePassword>(context, id, op, data).await
527 UserOperation::GetReportCount => {
528 do_websocket_operation::<GetReportCount>(context, id, op, data).await
530 UserOperation::GetUnreadCount => {
531 do_websocket_operation::<GetUnreadCount>(context, id, op, data).await
533 UserOperation::VerifyEmail => {
534 do_websocket_operation::<VerifyEmail>(context, id, op, data).await
537 // Private Message ops
538 UserOperation::MarkPrivateMessageAsRead => {
539 do_websocket_operation::<MarkPrivateMessageAsRead>(context, id, op, data).await
541 UserOperation::CreatePrivateMessageReport => {
542 do_websocket_operation::<CreatePrivateMessageReport>(context, id, op, data).await
544 UserOperation::ResolvePrivateMessageReport => {
545 do_websocket_operation::<ResolvePrivateMessageReport>(context, id, op, data).await
547 UserOperation::ListPrivateMessageReports => {
548 do_websocket_operation::<ListPrivateMessageReports>(context, id, op, data).await
552 UserOperation::GetModlog => do_websocket_operation::<GetModlog>(context, id, op, data).await,
553 UserOperation::PurgePerson => {
554 do_websocket_operation::<PurgePerson>(context, id, op, data).await
556 UserOperation::PurgeCommunity => {
557 do_websocket_operation::<PurgeCommunity>(context, id, op, data).await
559 UserOperation::PurgePost => do_websocket_operation::<PurgePost>(context, id, op, data).await,
560 UserOperation::PurgeComment => {
561 do_websocket_operation::<PurgeComment>(context, id, op, data).await
563 UserOperation::TransferCommunity => {
564 do_websocket_operation::<TransferCommunity>(context, id, op, data).await
566 UserOperation::LeaveAdmin => do_websocket_operation::<LeaveAdmin>(context, id, op, data).await,
569 UserOperation::FollowCommunity => {
570 do_websocket_operation::<FollowCommunity>(context, id, op, data).await
572 UserOperation::BlockCommunity => {
573 do_websocket_operation::<BlockCommunity>(context, id, op, data).await
575 UserOperation::BanFromCommunity => {
576 do_websocket_operation::<BanFromCommunity>(context, id, op, data).await
578 UserOperation::AddModToCommunity => {
579 do_websocket_operation::<AddModToCommunity>(context, id, op, data).await
583 UserOperation::LockPost => do_websocket_operation::<LockPost>(context, id, op, data).await,
584 UserOperation::FeaturePost => {
585 do_websocket_operation::<FeaturePost>(context, id, op, data).await
587 UserOperation::CreatePostLike => {
588 do_websocket_operation::<CreatePostLike>(context, id, op, data).await
590 UserOperation::MarkPostAsRead => {
591 do_websocket_operation::<MarkPostAsRead>(context, id, op, data).await
593 UserOperation::SavePost => do_websocket_operation::<SavePost>(context, id, op, data).await,
594 UserOperation::CreatePostReport => {
595 do_websocket_operation::<CreatePostReport>(context, id, op, data).await
597 UserOperation::ListPostReports => {
598 do_websocket_operation::<ListPostReports>(context, id, op, data).await
600 UserOperation::ResolvePostReport => {
601 do_websocket_operation::<ResolvePostReport>(context, id, op, data).await
603 UserOperation::GetSiteMetadata => {
604 do_websocket_operation::<GetSiteMetadata>(context, id, op, data).await
608 UserOperation::SaveComment => {
609 do_websocket_operation::<SaveComment>(context, id, op, data).await
611 UserOperation::CreateCommentLike => {
612 do_websocket_operation::<CreateCommentLike>(context, id, op, data).await
614 UserOperation::DistinguishComment => {
615 do_websocket_operation::<DistinguishComment>(context, id, op, data).await
617 UserOperation::CreateCommentReport => {
618 do_websocket_operation::<CreateCommentReport>(context, id, op, data).await
620 UserOperation::ListCommentReports => {
621 do_websocket_operation::<ListCommentReports>(context, id, op, data).await
623 UserOperation::ResolveCommentReport => {
624 do_websocket_operation::<ResolveCommentReport>(context, id, op, data).await
629 async fn do_websocket_operation<'a, 'b, Data>(
630 context: ContextData<LemmyContext>,
634 ) -> result::Result<String, LemmyError>
636 Data: Perform + SendActivity<Response = <Data as Perform>::Response> + Send,
637 for<'de> Data: Deserialize<'de>,
639 let parsed_data: Data = serde_json::from_value(data)?;
640 let res = parsed_data
641 .perform(&web::Data::new(context.deref().clone()), Some(id))
643 SendActivity::send_activity(&parsed_data, &res, &context).await?;
644 serialize_websocket_message(&op, &res)