Some more clean ups

src/helpers.rs

···
 use anyhow::anyhow;
 use axum::body::{Body, to_bytes};
 use axum::extract::Request;
-use axum::http::header::{CONTENT_LENGTH, CONTENT_TYPE};
+use axum::http::header::CONTENT_TYPE;
 use axum::http::{HeaderMap, StatusCode, Uri};
 use axum::response::{IntoResponse, Response};
 use axum_template::TemplateEngine;

src/main.rs

···
 use tower_governor::governor::GovernorConfigBuilder;
 use tower_http::compression::CompressionLayer;
 use tower_http::cors::{Any, CorsLayer};
-use tracing::error;
+use tracing::log;
 use tracing_subscriber::{EnvFilter, fmt, prelude::*};
 
 pub mod helpers;
···
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     setup_tracing();
-    //TODO may need to change where this reads from? Like an env variable for it's location?
+    //TODO may need to change where this reads from? Like an env variable for it's location? Or arg?
     dotenvy::from_path(Path::new("./pds.env"))?;
     let pds_root = env::var("PDS_DATA_DIRECTORY")?;
     let account_db_url = format!("{pds_root}/account.sqlite");
 
     let account_options = SqliteConnectOptions::new()
-        .journal_mode(SqliteJournalMode::Wal)
-        .filename(account_db_url);
+        .filename(account_db_url)
+        .busy_timeout(Duration::from_secs(5));
 
     let account_pool = SqlitePoolOptions::new()
         .max_connections(5)
···
     let options = SqliteConnectOptions::new()
         .journal_mode(SqliteJournalMode::Wal)
         .filename(bells_db_url)
-        .create_if_missing(true);
+        .create_if_missing(true)
+        .busy_timeout(Duration::from_secs(5));
     let pds_gatekeeper_pool = SqlitePoolOptions::new()
         .max_connections(5)
         .connect_with(options)
···
     //TODO add an override to manually load in the hbs templates
     let _ = hbs.register_embed_templates::<EmailTemplates>();
 
+    let pds_base_url =
+        env::var("PDS_BASE_URL").unwrap_or_else(|_| "http://localhost:3000".to_string());
+
     let state = AppState {
         account_pool,
         pds_gatekeeper_pool,
         reverse_proxy_client: client,
-        //TODO should be env prob
-        pds_base_url: "http://localhost:3000".to_string(),
+        pds_base_url,
         mailer,
         mailer_from: sent_from,
         template_engine: Engine::from(hbs),
···
 
     // Rate limiting
     //Allows 5 within 60 seconds, and after 60 should drop one off? So hit 5, then goes to 4 after 60 seconds.
-    let governor_conf = GovernorConfigBuilder::default()
+    let create_session_governor_conf = GovernorConfigBuilder::default()
+        .per_second(60)
+        .burst_size(5)
+        .finish()
+        .expect("failed to create governor config. this should not happen and is a bug");
+
+    // Create a second config with the same settings for the other endpoint
+    let sign_in_governor_conf = GovernorConfigBuilder::default()
         .per_second(60)
         .burst_size(5)
         .finish()
         .expect("failed to create governor config. this should not happen and is a bug");
 
-    let governor_limiter = governor_conf.limiter().clone();
+    let create_session_governor_limiter = create_session_governor_conf.limiter().clone();
+    let sign_in_governor_limiter = sign_in_governor_conf.limiter().clone();
     let interval = Duration::from_secs(60);
     // a separate background task to clean up
     std::thread::spawn(move || {
         loop {
             std::thread::sleep(interval);
-            tracing::info!("rate limiting storage size: {}", governor_limiter.len());
-            governor_limiter.retain_recent();
+            create_session_governor_limiter.retain_recent();
+            sign_in_governor_limiter.retain_recent();
         }
     });
 
···
         )
         .route(
             "/@atproto/oauth-provider/~api/sign-in",
-            post(sign_in), // .layer(GovernorLayer::new(governor_conf.clone()))),
+            post(sign_in).layer(GovernorLayer::new(sign_in_governor_conf)),
         )
         .route(
             "/xrpc/com.atproto.server.createSession",
-            post(create_session.layer(GovernorLayer::new(governor_conf))),
+            post(create_session.layer(GovernorLayer::new(create_session_governor_conf))),
         )
         .layer(CompressionLayer::new())
         .layer(cors)
         .with_state(state);
 
-    let host = env::var("HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
-    let port: u16 = env::var("PORT")
+    let host = env::var("GATEKEEPER_HOST").unwrap_or_else(|_| "127.0.0.1".to_string());
+    let port: u16 = env::var("GATEKEEPER_PORT")
         .ok()
         .and_then(|s| s.parse().ok())
         .unwrap_or(8080);
···
     .with_graceful_shutdown(shutdown_signal());
 
     if let Err(err) = server.await {
-        error!(error = %err, "server error");
+        log::error!("server error:{err}");
     }
 
     Ok(())

src/oauth_provider.rs

···
 use serde::{Deserialize, Serialize};
 use tracing::log;
 
-fn is_disallowed_header(name: &HeaderName) -> bool {
-    // RFC 7230 hop-by-hop headers and other problematic ones for proxying a new body
-    // Use lowercase comparison; HeaderName equality is case-insensitive but we compare by string for a set
-    match name.as_str() {
-        // Hop-by-hop
-        "connection"
-        | "keep-alive"
-        | "proxy-authenticate"
-        | "proxy-authorization"
-        | "te"
-        | "trailer"
-        | "transfer-encoding"
-        | "upgrade" => true,
-        // Payload or routing related we should not forward
-        "host" | "content-length" | "content-encoding" | "expect" => true,
-        // Compression negotiation can interfere; let upstream decide defaults
-        // We can drop Accept-Encoding to avoid getting compressed payloads if not needed
-        "accept-encoding" => true,
-        _ => false,
-    }
-}
-
-fn copy_filtered_headers(src: &HeaderMap, dst: &mut HeaderMap) {
-    for (name, value) in src.iter() {
-        if is_disallowed_header(name) {
-            continue;
-        }
-        // Only copy valid headers
-        if let Ok(hv) = HeaderValue::from_bytes(value.as_bytes()) {
-            dst.insert(name.clone(), hv);
-        }
-    }
-}
-
 #[derive(Serialize, Deserialize, Clone)]
 pub struct SignInRequest {
     pub username: String,
···
     let password = payload.password.clone();
     let auth_factor_token = payload.email_otp.clone();
 
-    //TODO need to pass in a flag to ignore app passwords for Oauth
-    // Run the shared pre-auth logic to validate and check 2FA requirement
     match preauth_check(&state, &identifier, &password, auth_factor_token, true).await {
         Ok(result) => match result {
             AuthResult::WrongIdentityOrPassword => oauth_json_error_response(
···
                 );
 
                 let mut req = axum::http::Request::post(uri);
-                // if let Some(cookie) = headers.get("Cookie") {
-                //     req = req.header("Cookie", cookie.to_str().unwrap());
-                // }
                 if let Some(req_headers) = req.headers_mut() {
-                    // Copy headers but remove hop-by-hop and problematic ones
+                    // Copy headers but remove problematic ones. There was an issue with the PDS not parsing the body fully if i forwarded all headers
                     copy_filtered_headers(&headers, req_headers);
-                    // Ensure JSON content type is set explicitly
+                    //Setting the content type to application/json manually
                     req_headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
                 }
 
                 payload.email_otp = None;
-                // let payload_bytes =
-                //     serde_json::to_vec(&payload).map_err(|_| StatusCode::BAD_REQUEST)?;
-                let body = serde_json::to_string(&payload).map_err(|_| StatusCode::BAD_REQUEST)?;
+                let payload_bytes =
+                    serde_json::to_vec(&payload).map_err(|_| StatusCode::BAD_REQUEST)?;
 
                 let req = req
-                    .body(Body::from(body))
+                    .body(Body::from(payload_bytes))
                     .map_err(|_| StatusCode::BAD_REQUEST)?;
 
                 let proxied = state
···
         }
     }
 }
+
+fn is_disallowed_header(name: &HeaderName) -> bool {
+    // possible problematic headers with proxying
+    matches!(
+        name.as_str(),
+        "connection"
+            | "keep-alive"
+            | "proxy-authenticate"
+            | "proxy-authorization"
+            | "te"
+            | "trailer"
+            | "transfer-encoding"
+            | "upgrade"
+            | "host"
+            | "content-length"
+            | "content-encoding"
+            | "expect"
+            | "accept-encoding"
+    )
+}
+
+fn copy_filtered_headers(src: &HeaderMap, dst: &mut HeaderMap) {
+    for (name, value) in src.iter() {
+        if is_disallowed_header(name) {
+            continue;
+        }
+        // Only copy valid headers
+        if let Ok(hv) = HeaderValue::from_bytes(value.as_bytes()) {
+            dst.insert(name.clone(), hv);
+        }
+    }
+}

src/xrpc/com_atproto_server.rs

···
 use serde::{Deserialize, Serialize};
 use serde_json;
 use tracing::log;
-use tracing::log::log;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
···
 pub async fn create_session(
     State(state): State<AppState>,
     headers: HeaderMap,
-    Json(mut payload): extract::Json<CreateSessionRequest>,
+    Json(payload): extract::Json<CreateSessionRequest>,
 ) -> Result<Response<Body>, StatusCode> {
     let identifier = payload.identifier.clone();
     let password = payload.password.clone();