use std::sync::Arc;
use std::time::Instant;

use salvo::http::ParseError;
use salvo::http::StatusCode;
use salvo::prelude::*;
use serde::Serialize;
use serde_json::Value;
use uuid::Uuid;

use crate::AppState;
use crate::config::AccessLogConfig;
use crate::metrics;
use crate::models::{Notification, NotificationContext, NotifyResponse};

pub const MAX_REQUEST_SIZE: usize = 512 * 1024;

pub fn build_router(state: Arc<AppState>) -> Router {
    Router::with_hoop(affix_state::inject(state))
        .push(Router::with_path("_matrix/push/v1/notify").post(notify))
        .push(Router::with_path("health").get(health))
}

pub fn build_router_with_access_log(state: Arc<AppState>, access_log: &AccessLogConfig) -> Router {
    let use_forwarded_for = access_log.x_forwarded_for;
    Router::with_hoop(affix_state::inject(state))
        .hoop(AccessLogger { use_forwarded_for })
        .push(Router::with_path("_matrix/push/v1/notify").post(notify))
        .push(Router::with_path("health").get(health))
}

struct AccessLogger {
    use_forwarded_for: bool,
}

#[handler]
impl AccessLogger {
    async fn handle(
        &self,
        req: &mut Request,
        depot: &mut Depot,
        res: &mut Response,
        ctrl: &mut FlowCtrl,
    ) {
        let method = req.method().clone();
        let path = req.uri().path().to_owned();
        let remote_addr = if self.use_forwarded_for {
            req.header::<String>("x-forwarded-for")
                .unwrap_or_else(|| req.remote_addr().to_string())
        } else {
            req.remote_addr().to_string()
        };
        let started = Instant::now();

        ctrl.call_next(req, depot, res).await;

        let status = res.status_code.unwrap_or(StatusCode::OK).as_u16();
        let elapsed = started.elapsed();

        // Skip noisy health check logging at info level
        if path == "/health" {
            tracing::debug!(
                method = %method,
                path = %path,
                status = status,
                duration_ms = elapsed.as_millis() as u64,
                remote = %remote_addr,
                "request handled"
            );
        } else {
            tracing::info!(
                method = %method,
                path = %path,
                status = status,
                duration_ms = elapsed.as_millis() as u64,
                remote = %remote_addr,
                "request handled"
            );
        }
    }
}

#[handler]
async fn health(res: &mut Response) {
    res.status_code(StatusCode::OK);
    res.render(Text::Plain(""));
}

#[handler]
async fn notify(req: &mut Request, depot: &mut Depot, res: &mut Response) {
    let started = Instant::now();
    let _inflight = metrics::track_inflight("V1NotifyHandler");

    let state = match depot.obtain::<Arc<AppState>>() {
        Ok(state) => state.clone(),
        Err(_) => {
            finish_text(
                res,
                StatusCode::INTERNAL_SERVER_ERROR,
                "application state missing",
                started,
            );
            return;
        }
    };

    let body = match req.payload_with_max_size(MAX_REQUEST_SIZE).await {
        Ok(bytes) => bytes,
        Err(ParseError::PayloadTooLarge) => {
            finish_text(
                res,
                StatusCode::PAYLOAD_TOO_LARGE,
                "request body exceeds 512 KiB",
                started,
            );
            return;
        }
        Err(error) => {
            tracing::warn!(error = %error, "failed to read request body");
            finish_text(
                res,
                StatusCode::BAD_REQUEST,
                "failed to read request body",
                started,
            );
            return;
        }
    };

    let raw = match serde_json::from_slice::<Value>(body) {
        Ok(raw) => raw,
        Err(error) => {
            tracing::warn!(error = %error, "expected JSON request body");
            finish_text(
                res,
                StatusCode::BAD_REQUEST,
                "expected JSON request body",
                started,
            );
            return;
        }
    };

    let Some(notification_value) = raw.get("notification").cloned() else {
        finish_text(
            res,
            StatusCode::BAD_REQUEST,
            "missing notification field",
            started,
        );
        return;
    };

    if !notification_value.is_object() {
        finish_text(
            res,
            StatusCode::BAD_REQUEST,
            "notification must be an object",
            started,
        );
        return;
    }

    let notification: Notification = match serde_json::from_value(notification_value) {
        Ok(notification) => notification,
        Err(error) => {
            tracing::warn!(error = %error, "invalid notification payload");
            finish_text(
                res,
                StatusCode::BAD_REQUEST,
                "invalid notification payload",
                started,
            );
            return;
        }
    };
    metrics::notification_received();

    if notification.devices.is_empty() {
        finish_text(
            res,
            StatusCode::BAD_REQUEST,
            "no devices in notification",
            started,
        );
        return;
    }

    let context = NotificationContext {
        request_id: Uuid::new_v4().to_string(),
        start_time: Instant::now(),
    };

    let mut rejected = Vec::new();
    for device in &notification.devices {
        metrics::device_push_received();
        let pushkins = state.registry.find_pushkins(&device.app_id);
        match pushkins.as_slice() {
            [] => {
                tracing::warn!(request_id = %context.request_id, app_id = %device.app_id, pushkey = %device.pushkey, "unknown app id");
                rejected.push(device.pushkey.clone());
                continue;
            }
            [pushkin] => {
                metrics::pushkin_selected(pushkin.name());
                match pushkin
                    .dispatch_notification(&notification, device, &context)
                    .await
                {
                    Ok(mut pushkin_rejected) => rejected.append(&mut pushkin_rejected),
                    Err(error) if error.is_remote() => {
                        tracing::warn!(error = %error, request_id = %context.request_id, "remote dispatch failure");
                        finish_text(res, StatusCode::BAD_GATEWAY, &error.to_string(), started);
                        return;
                    }
                    Err(error) => {
                        tracing::error!(error = %error, request_id = %context.request_id, "internal dispatch failure");
                        finish_text(
                            res,
                            StatusCode::INTERNAL_SERVER_ERROR,
                            &error.to_string(),
                            started,
                        );
                        return;
                    }
                }
            }
            _ => {
                tracing::warn!(request_id = %context.request_id, app_id = %device.app_id, pushkey = %device.pushkey, "ambiguous app id");
                rejected.push(device.pushkey.clone());
            }
        }
    }

    finish_json(res, StatusCode::OK, NotifyResponse { rejected }, started);
}

fn finish_text(res: &mut Response, status: StatusCode, body: &str, started: Instant) {
    res.status_code(status);
    res.render(Text::Plain(body.to_owned()));
    metrics::pushgateway_response(status);
    metrics::observe_notify_handle(status, started.elapsed());
}

fn finish_json<T: Serialize + Send>(
    res: &mut Response,
    status: StatusCode,
    body: T,
    started: Instant,
) {
    res.status_code(status);
    res.render(Json(body));
    metrics::pushgateway_response(status);
    metrics::observe_notify_handle(status, started.elapsed());
}

#[cfg(test)]
mod tests {
    use std::collections::HashMap;
    use std::sync::Arc;
    use std::time::Duration;

    use async_trait::async_trait;
    use salvo::test::{ResponseExt, TestClient};
    use serde_json::json;
    use tokio::time::sleep;

    use super::*;
    use crate::error::DispatchError;
    use crate::models::{Device, Notification, NotificationContext};
    use crate::pushkin::{AppMatcher, ConcurrencyGate, Pushkin, PushkinRegistry};

    #[derive(Debug, Clone)]
    enum TestBehavior {
        Accept,
        Reject,
        RemoteError,
        InternalError,
        SlowAccept,
    }

    struct TestPushkin {
        matcher: AppMatcher,
        gate: Option<ConcurrencyGate>,
        behavior: TestBehavior,
    }

    impl TestPushkin {
        fn new(name: &str, behavior: TestBehavior) -> Self {
            Self {
                matcher: AppMatcher::new(name.to_owned()).unwrap(),
                gate: None,
                behavior,
            }
        }

        fn with_limit(name: &str, behavior: TestBehavior, limit: usize) -> Self {
            Self {
                matcher: AppMatcher::new(name.to_owned()).unwrap(),
                gate: Some(ConcurrencyGate::new(limit)),
                behavior,
            }
        }
    }

    #[async_trait]
    impl Pushkin for TestPushkin {
        fn name(&self) -> &str {
            self.matcher.name()
        }

        fn handles_appid(&self, appid: &str) -> bool {
            self.matcher.handles_appid(appid)
        }

        async fn dispatch_notification(
            &self,
            _notification: &Notification,
            device: &Device,
            _context: &NotificationContext,
        ) -> Result<Vec<String>, DispatchError> {
            let _permit = self
                .gate
                .as_ref()
                .map(|gate| gate.acquire(self.name()))
                .transpose()?;

            match self.behavior {
                TestBehavior::Accept => Ok(vec![]),
                TestBehavior::Reject => Ok(vec![device.pushkey.clone()]),
                TestBehavior::RemoteError => Err(DispatchError::remote("synthetic remote failure")),
                TestBehavior::InternalError => {
                    Err(DispatchError::internal("synthetic internal failure"))
                }
                TestBehavior::SlowAccept => {
                    sleep(Duration::from_millis(100)).await;
                    Ok(vec![])
                }
            }
        }
    }

    fn test_service(pushkins: Vec<(&str, Arc<dyn Pushkin>)>) -> Service {
        let registry = PushkinRegistry::new(
            pushkins
                .into_iter()
                .map(|(name, pushkin)| (name.to_owned(), pushkin))
                .collect::<HashMap<_, _>>(),
        );
        let state = Arc::new(AppState::new(Arc::new(registry)));
        Service::new(build_router(state))
    }

    fn payload(devices: Vec<Value>) -> Value {
        json!({
            "notification": {
                "event_id": "$event",
                "room_id": "!room:example.com",
                "sender": "@alice:example.com",
                "sender_display_name": "Alice",
                "type": "m.room.message",
                "content": {
                    "msgtype": "m.text",
                    "body": "hello"
                },
                "devices": devices
            }
        })
    }

    fn device(app_id: &str, pushkey: &str) -> Value {
        json!({
            "app_id": app_id,
            "pushkey": pushkey,
            "pushkey_ts": 42
        })
    }

    #[tokio::test]
    async fn accepted_devices_are_not_rejected() {
        let service = test_service(vec![(
            "com.example.app",
            Arc::new(TestPushkin::new("com.example.app", TestBehavior::Accept)),
        )]);

        let mut response = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .json(&payload(vec![device("com.example.app", "accept")]))
            .send(&service)
            .await;

        assert_eq!(response.status_code.unwrap(), StatusCode::OK);
        assert_eq!(
            response.take_json::<NotifyResponse>().await.unwrap(),
            NotifyResponse { rejected: vec![] }
        );
    }

    #[tokio::test]
    async fn rejected_devices_are_reported() {
        let service = test_service(vec![(
            "com.example.app",
            Arc::new(TestPushkin::new("com.example.app", TestBehavior::Reject)),
        )]);

        let mut response = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .json(&payload(vec![device("com.example.app", "reject")]))
            .send(&service)
            .await;

        assert_eq!(response.status_code.unwrap(), StatusCode::OK);
        assert_eq!(
            response.take_json::<NotifyResponse>().await.unwrap(),
            NotifyResponse {
                rejected: vec!["reject".to_owned()]
            }
        );
    }

    #[tokio::test]
    async fn ambiguous_app_ids_are_rejected() {
        let service = test_service(vec![
            (
                "*.example.*",
                Arc::new(TestPushkin::new("*.example.*", TestBehavior::Accept)),
            ),
            (
                "com.example.a*",
                Arc::new(TestPushkin::new("com.example.a*", TestBehavior::Accept)),
            ),
        ]);

        let mut response = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .json(&payload(vec![device("com.example.app", "spqr")]))
            .send(&service)
            .await;

        assert_eq!(response.status_code.unwrap(), StatusCode::OK);
        assert_eq!(
            response.take_json::<NotifyResponse>().await.unwrap(),
            NotifyResponse {
                rejected: vec!["spqr".to_owned()]
            }
        );
    }

    #[tokio::test]
    async fn remote_errors_map_to_502() {
        let service = test_service(vec![(
            "com.example.app",
            Arc::new(TestPushkin::new(
                "com.example.app",
                TestBehavior::RemoteError,
            )),
        )]);

        let response = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .json(&payload(vec![device("com.example.app", "remote")]))
            .send(&service)
            .await;

        assert_eq!(response.status_code.unwrap(), StatusCode::BAD_GATEWAY);
    }

    #[tokio::test]
    async fn internal_errors_map_to_500() {
        let service = test_service(vec![(
            "com.example.app",
            Arc::new(TestPushkin::new(
                "com.example.app",
                TestBehavior::InternalError,
            )),
        )]);

        let response = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .json(&payload(vec![device("com.example.app", "boom")]))
            .send(&service)
            .await;

        assert_eq!(
            response.status_code.unwrap(),
            StatusCode::INTERNAL_SERVER_ERROR
        );
    }

    #[tokio::test]
    async fn oversized_requests_are_rejected() {
        let service = test_service(vec![(
            "com.example.app",
            Arc::new(TestPushkin::new("com.example.app", TestBehavior::Accept)),
        )]);

        let response = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .text("x".repeat(MAX_REQUEST_SIZE + 1))
            .send(&service)
            .await;

        assert_eq!(response.status_code.unwrap(), StatusCode::PAYLOAD_TOO_LARGE);
    }

    #[tokio::test]
    async fn per_pushkin_concurrency_limit_returns_502() {
        let service = test_service(vec![(
            "com.example.app",
            Arc::new(TestPushkin::with_limit(
                "com.example.app",
                TestBehavior::SlowAccept,
                1,
            )),
        )]);

        let request_a = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .json(&payload(vec![device("com.example.app", "one")]))
            .send(&service);
        let request_b = TestClient::post("http://127.0.0.1/_matrix/push/v1/notify")
            .json(&payload(vec![device("com.example.app", "two")]))
            .send(&service);

        let (response_a, response_b) = tokio::join!(request_a, request_b);
        let statuses = [
            response_a.status_code.unwrap(),
            response_b.status_code.unwrap(),
        ];

        assert!(statuses.contains(&StatusCode::OK));
        assert!(statuses.contains(&StatusCode::BAD_GATEWAY));
    }
}
