//! E2E resilience tests: retry policies, error handling, edge cases.
//!
//! SOP Categories E (Retry & Resilience) and F (Error Handling & Edge Cases).

mod common;

use std::net::SocketAddr;
use std::sync::atomic::{AtomicU32, Ordering};

use async_trait::async_trait;

use knafeh::client::retry::RetryPolicy;
use knafeh::codec::JsonCodec;
use knafeh::error::{KnafehError, RpcStatusCode};
use knafeh::rpc::message::{RpcRequest, RpcResponse};
use knafeh::rpc::service::{MethodDescriptor, MethodKind, Service};
use knafeh::rpc::stream::{RpcStreamRequest, RpcStreamResponse};
use knafeh::{Client, Server};

// ---------------------------------------------------------------------------
// TLS certs (shared with e2e_test.rs pattern)
// ---------------------------------------------------------------------------

use common::{start_test_server, tls_server};

// ---------------------------------------------------------------------------
// Test services
// ---------------------------------------------------------------------------

/// Returns Unavailable for the first N calls, then Ok.
/// The counter is shared across all requests to this service instance.
struct FailService {
    fail_count: u32,
    call_counter: AtomicU32,
}

impl FailService {
    fn new(fail_count: u32) -> Self {
        Self {
            fail_count,
            call_counter: AtomicU32::new(0),
        }
    }
}

#[async_trait]
impl Service for FailService {
    fn name(&self) -> &str {
        "fail"
    }

    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![MethodDescriptor {
            name: "call".to_string(),
            kind: MethodKind::Unary,
        }]
    }

    async fn call_unary(
        &self,
        _method: &str,
        request: RpcRequest,
    ) -> Result<RpcResponse, KnafehError> {
        let n = self.call_counter.fetch_add(1, Ordering::SeqCst);
        if n < self.fail_count {
            Err(KnafehError::Service {
                code: RpcStatusCode::Unavailable,
                message: format!("temporarily unavailable (attempt {})", n + 1),
            })
        } else {
            Ok(RpcResponse::ok(request.body))
        }
    }

    async fn call_server_stream(
        &self,
        _: &str,
        _: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}

/// Returns the specified RPC status code.
struct ErrorService;

#[async_trait]
impl Service for ErrorService {
    fn name(&self) -> &str {
        "error"
    }

    fn methods(&self) -> Vec<MethodDescriptor> {
        vec![
            MethodDescriptor {
                name: "not_found".to_string(),
                kind: MethodKind::Unary,
            },
            MethodDescriptor {
                name: "permission_denied".to_string(),
                kind: MethodKind::Unary,
            },
            MethodDescriptor {
                name: "internal".to_string(),
                kind: MethodKind::Unary,
            },
            MethodDescriptor {
                name: "invalid_argument".to_string(),
                kind: MethodKind::Unary,
            },
        ]
    }

    async fn call_unary(
        &self,
        method: &str,
        _request: RpcRequest,
    ) -> Result<RpcResponse, KnafehError> {
        let (code, msg) = match method {
            "not_found" => (RpcStatusCode::NotFound, "resource not found"),
            "permission_denied" => (RpcStatusCode::PermissionDenied, "access denied"),
            "internal" => (RpcStatusCode::Internal, "internal server error"),
            "invalid_argument" => (RpcStatusCode::InvalidArgument, "bad request"),
            _ => (RpcStatusCode::Unknown, "unknown"),
        };
        Ok(RpcResponse::error(code, msg))
    }

    async fn call_server_stream(
        &self,
        _: &str,
        _: RpcRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_client_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcResponse, KnafehError> {
        unimplemented!()
    }

    async fn call_bidi_stream(
        &self,
        _: &str,
        _: RpcStreamRequest,
    ) -> Result<RpcStreamResponse, KnafehError> {
        unimplemented!()
    }
}

// ===========================================================================
// Category E: Retry & Resilience
// ===========================================================================

/// E1: Retry on Unavailable — service fails first 2 calls then succeeds.
#[tokio::test]
async fn test_e2e_retry_on_unavailable() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(FailService::new(2))
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;

    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .retry(
            RetryPolicy::exponential(3)
                .with_initial_backoff(std::time::Duration::from_millis(10))
                .with_jitter(false),
        )
        .build()
        .await
        .unwrap();

    // Should succeed after retrying past the 2 failures.
    let resp = client
        .call("fail/call", br#"{"ok":true}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Ok);
    assert_eq!(resp.body, br#"{"ok":true}"#);
}

/// E2: Non-retryable error — InvalidArgument is not retried.
#[tokio::test]
async fn test_e2e_non_retryable_error() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(ErrorService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;

    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .retry(RetryPolicy::exponential(3))
        .build()
        .await
        .unwrap();

    // InvalidArgument is non-retryable — should return immediately.
    let resp = client
        .call("error/invalid_argument", br#"{}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::InvalidArgument);
}

/// E3: Retry exhaustion — service always fails, retries run out.
#[tokio::test]
async fn test_e2e_retry_exhaustion() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(FailService::new(100)) // always fails
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;

    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .retry(
            RetryPolicy::exponential(2)
                .with_initial_backoff(std::time::Duration::from_millis(10))
                .with_jitter(false),
        )
        .build()
        .await
        .unwrap();

    // Should fail after 3 total attempts (1 initial + 2 retries).
    // The final attempt returns the Unavailable response as-is.
    let resp = client.call("fail/call", br#"{}"#.to_vec()).await.unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Unavailable);
}

/// E4: Backoff timing — each retry waits longer than the previous.
#[tokio::test]
async fn test_e2e_backoff_timing() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(FailService::new(100)) // always fails
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;

    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .retry(
            RetryPolicy::exponential(2)
                .with_initial_backoff(std::time::Duration::from_millis(50))
                .with_jitter(false),
        )
        .build()
        .await
        .unwrap();

    let start = std::time::Instant::now();
    let _ = client.call("fail/call", br#"{}"#.to_vec()).await;
    let elapsed = start.elapsed();

    // 2 retries with 50ms initial backoff (2x multiplier, no jitter):
    // ~50ms + ~100ms = ~150ms minimum.
    assert!(
        elapsed >= std::time::Duration::from_millis(100),
        "expected >= 100ms, got {:?}",
        elapsed
    );
}

// ===========================================================================
// Category F: Error Handling & Edge Cases
// ===========================================================================

/// F1: Codec error — server returns InvalidArgument for non-JSON body.
#[tokio::test]
async fn test_e2e_codec_error() {
    // EchoService echoes body, but the codec validates JSON.
    // Sending invalid JSON should fail at the client codec.encode() step.
    let (addr, _handle) = start_test_server(
        Server::builder()
            .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
            .tls(tls_server())
            .add_service(ErrorService)
            .build()
            .unwrap(),
    )
    .await;
    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .build()
        .await
        .unwrap();

    // Client-side codec should reject this before it even hits the wire.
    let err = client
        .call("error/internal", b"not json {{{".to_vec())
        .await
        .unwrap_err();
    assert!(matches!(err, KnafehError::Codec(_)));
}

/// F3: Service error propagation — various status codes.
#[tokio::test]
async fn test_e2e_service_error_propagation() {
    let server = Server::builder()
        .bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        .tls(tls_server())
        .codec(JsonCodec::new())
        .add_service(ErrorService)
        .build()
        .unwrap();
    let (addr, _handle) = start_test_server(server).await;
    let client = Client::builder()
        .endpoint(addr.to_string())
        .codec(JsonCodec::new())
        .build()
        .await
        .unwrap();

    // NotFound
    let resp = client
        .call("error/not_found", br#"{}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::NotFound);
    assert_eq!(resp.status.message, "resource not found");

    // PermissionDenied
    let resp = client
        .call("error/permission_denied", br#"{}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::PermissionDenied);
    assert_eq!(resp.status.message, "access denied");

    // Internal
    let resp = client
        .call("error/internal", br#"{}"#.to_vec())
        .await
        .unwrap();
    assert_eq!(resp.status.code, RpcStatusCode::Internal);
    assert_eq!(resp.status.message, "internal server error");
}
