Remove SendActivity and Perform traits, rely on channel (#3596)

[lemmy.git] / src / lib.rs

diff --git a/src/lib.rs b/src/lib.rs
index b6fd64ec8096c0640549774339bf11bbc1553cbc..e07ae2685d478736c756b320de429bd2f1914e5c 100644 (file)
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -10,17 +10,28 @@ pub mod telemetry;
 use crate::{code_migrations::run_advanced_migrations, root_span_builder::QuieterRootSpanBuilder};
 use activitypub_federation::config::{FederationConfig, FederationMiddleware};
 use actix_cors::Cors;
-use actix_web::{middleware, web::Data, App, HttpServer, Result};
+use actix_web::{
+  middleware::{self, ErrorHandlers},
+  web::Data,
+  App,
+  HttpServer,
+  Result,
+};
 use lemmy_api_common::{
   context::LemmyContext,
   lemmy_db_views::structs::SiteView,
   request::build_user_agent,
+  send_activity::MATCH_OUTGOING_ACTIVITIES,
   utils::{
     check_private_instance_and_federation_enabled,
     local_site_rate_limit_to_rate_limit_config,
   },
 };
-use lemmy_apub::{VerifyUrlData, FEDERATION_HTTP_FETCH_LIMIT};
+use lemmy_apub::{
+  activities::{handle_outgoing_activities, match_outgoing_activities},
+  VerifyUrlData,
+  FEDERATION_HTTP_FETCH_LIMIT,
+};
 use lemmy_db_schema::{
   source::secret::Secret,
   utils::{build_db_pool, get_database_url, run_migrations},
@@ -29,6 +40,7 @@ use lemmy_routes::{feeds, images, nodeinfo, webfinger};
 use lemmy_utils::{
   error::LemmyError,
   rate_limit::RateLimitCell,
+  response::jsonify_plain_text_errors,
   settings::SETTINGS,
   SYNCHRONOUS_FEDERATION,
 };
@@ -68,15 +80,15 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
   let pool = build_db_pool(&settings).await?;
 
   // Run the Code-required migrations
-  run_advanced_migrations(&pool, &settings).await?;
+  run_advanced_migrations(&mut (&pool).into(), &settings).await?;
 
   // Initialize the secrets
-  let secret = Secret::init(&pool)
+  let secret = Secret::init(&mut (&pool).into())
     .await
     .expect("Couldn't initialize secrets.");
 
   // Make sure the local site is set up.
-  let site_view = SiteView::read_local(&pool)
+  let site_view = SiteView::read_local(&mut (&pool).into())
     .await
     .expect("local site not set up");
   let local_site = site_view.local_site;
@@ -146,7 +158,7 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
     .retry_count(settings.retry_count)
     .debug(*SYNCHRONOUS_FEDERATION)
     .http_signature_compat(true)
-    .url_verifier(Box::new(VerifyUrlData(context.pool().clone())))
+    .url_verifier(Box::new(VerifyUrlData(context.inner_pool().clone())))
     .build()
     .await?;
 
@@ -156,11 +168,19 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
   let prom_api_metrics = PrometheusMetricsBuilder::new("lemmy_api")
     .registry(default_registry().clone())
     .build()
-    .unwrap();
+    .expect("Should always be buildable");
+
+  MATCH_OUTGOING_ACTIVITIES
+    .set(Box::new(move |d, c| {
+      Box::pin(match_outgoing_activities(d, c))
+    }))
+    .expect("set function pointer");
+  let request_data = federation_config.to_request_data();
+  let outgoing_activities_task = tokio::task::spawn(handle_outgoing_activities(request_data));
 
   // Create Http server with websocket support
   HttpServer::new(move || {
-    let cors_origin = std::env::var("LEMMY_CORS_ORIGIN");
+    let cors_origin = env::var("LEMMY_CORS_ORIGIN");
     let cors_config = match (cors_origin, cfg!(debug_assertions)) {
       (Ok(origin), false) => Cors::default()
         .allowed_origin(&origin)
@@ -181,6 +201,7 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
       .wrap(middleware::Compress::default())
       .wrap(cors_config)
       .wrap(TracingLogger::<QuieterRootSpanBuilder>::new())
+      .wrap(ErrorHandlers::new().default_handler(jsonify_plain_text_errors))
       .app_data(Data::new(context.clone()))
       .app_data(Data::new(rate_limit_cell.clone()))
       .wrap(FederationMiddleware::new(federation_config.clone()));
@@ -205,6 +226,9 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
   .run()
   .await?;
 
+  // Wait for outgoing apub sends to complete
+  outgoing_activities_task.await??;
+
   Ok(())
 }