use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use std::time::Duration;

use tokio::io::AsyncReadExt;

use gitbaby::server::{
    AuthContext, AuthScheme, ChildPackReader, ChildPackWriter, EndpointKind, IngestReport,
    PackStats, Principal, RefUpdateBatch, RefUpdateEntry, ServerBackend, ServerError,
    ServerRequest, StreamGuard, StreamOutcome, buf_empty_reader, empty_reader,
};

static ENV_LOCK: Mutex<()> = Mutex::new(());

fn lock_or_recover() -> MutexGuard<'static, ()> {
    match ENV_LOCK.lock() {
        Ok(g) => g,
        Err(PoisonError { .. }) => panic!("env lock poisoned"),
    }
}

#[derive(Debug)]
struct PlaceholderBackend;

#[async_trait::async_trait]
impl ServerBackend for PlaceholderBackend {
    async fn authorize(&self, _req: &ServerRequest) -> Result<AuthContext, ServerError> {
        Err(ServerError::Unimplemented(
            "server::ServerBackend::authorize",
        ))
    }

    async fn list_refs_stream(
        &self,
        _req: &ServerRequest,
    ) -> Result<Box<dyn tokio::io::AsyncRead + Unpin + Send + Sync>, ServerError> {
        Err(ServerError::Unimplemented(
            "server::ServerBackend::list_refs_stream",
        ))
    }

    async fn produce_pack_stream(
        &self,
        _req: &ServerRequest,
        _wants: &[String],
    ) -> Result<Box<dyn tokio::io::AsyncRead + Unpin + Send + Sync>, ServerError> {
        Err(ServerError::Unimplemented(
            "server::ServerBackend::produce_pack_stream",
        ))
    }

    async fn ingest_pack_stream(
        &self,
        _req: &ServerRequest,
        _batch: RefUpdateBatch,
    ) -> Result<IngestReport, ServerError> {
        Err(ServerError::Unimplemented(
            "server::ServerBackend::ingest_pack_stream",
        ))
    }

    async fn lfs_get_stream(
        &self,
        _req: &ServerRequest,
        _oid: &str,
    ) -> Result<Box<dyn tokio::io::AsyncRead + Unpin + Send + Sync>, ServerError> {
        Err(ServerError::Unimplemented(
            "server::ServerBackend::lfs_get_stream",
        ))
    }

    async fn lfs_put_stream(
        &self,
        _req: &ServerRequest,
        _oid: &str,
        _declared_size: u64,
    ) -> Result<IngestReport, ServerError> {
        Err(ServerError::Unimplemented(
            "server::ServerBackend::lfs_put_stream",
        ))
    }
}

fn make_request(method: &str, path: &str, endpoint: EndpointKind) -> ServerRequest {
    ServerRequest::new(method, path, endpoint, "test-service", empty_reader())
}

#[test]
fn auth_scheme_str_round_trip() {
    let _guard = lock_or_recover();
    assert_eq!(AuthScheme::Anonymous.as_str(), "anonymous");
    assert_eq!(AuthScheme::Basic.as_str(), "basic");
    assert_eq!(AuthScheme::Bearer.as_str(), "bearer");
    assert_eq!(AuthScheme::Ssh.as_str(), "ssh");
    assert_eq!(AuthScheme::default(), AuthScheme::Anonymous);
}

#[test]
fn auth_context_anonymous_has_no_principal() {
    let ctx = AuthContext::anonymous();
    assert_eq!(ctx.scheme, AuthScheme::Anonymous);
    assert!(ctx.principal.is_none());
    assert!(ctx.raw.is_none());
}

#[test]
fn auth_context_with_principal_keeps_scheme_and_principal() {
    let ctx = AuthContext::with_principal(
        AuthScheme::Bearer,
        Principal::new("alice", AuthScheme::Bearer),
    );
    assert_eq!(ctx.scheme, AuthScheme::Bearer);
    let p = ctx.principal.expect("principal present");
    assert_eq!(p.name, "alice");
    assert_eq!(p.scheme, AuthScheme::Bearer);
}

#[test]
fn ref_update_batch_push_and_count() {
    let batch = RefUpdateBatch::new()
        .push(
            "refs/heads/main",
            "0000000000000000000000000000000000000000",
            None,
        )
        .push(
            "refs/heads/dev",
            "1111111111111111111111111111111111111111",
            Some("0000000000000000000000000000000000000000".to_string()),
        );

    assert_eq!(batch.len(), 2);
    assert!(!batch.is_empty());
    assert_eq!(batch.updates.len(), 2);
    assert_eq!(batch.updates[0].name, "refs/heads/main");
    assert!(batch.updates[0].old_oid.is_none());
    assert!(batch.updates[1].old_oid.is_some());
}

#[test]
fn stream_guard_cancel_is_observable_via_handle() {
    let guard = StreamGuard::new();
    let handle = guard.handle();
    assert!(!guard.is_cancelled());
    assert!(!handle.is_cancelled());

    guard.cancel();
    assert!(guard.is_cancelled());
    assert!(handle.is_cancelled());

    handle.cancel();
    assert!(guard.is_cancelled());
}

#[test]
fn server_request_builder_populates_fields() {
    let req = make_request("GET", "/info/refs", EndpointKind::InfoRefs)
        .with_query("service=git-upload-pack")
        .with_header(
            "Content-Type",
            "application/x-git-upload-pack-advertisement",
        )
        .with_auth(AuthContext::anonymous())
        .with_output_cap(1024)
        .with_remote_addr("127.0.0.1:9418");

    assert_eq!(req.method, "GET");
    assert_eq!(req.path, "/info/refs");
    assert_eq!(req.query.as_deref(), Some("service=git-upload-pack"));
    assert_eq!(
        req.headers.get("Content-Type").map(String::as_str),
        Some("application/x-git-upload-pack-advertisement")
    );
    assert_eq!(req.output_cap, 1024);
    assert_eq!(req.remote_addr.as_deref(), Some("127.0.0.1:9418"));
    assert_eq!(req.endpoint, EndpointKind::InfoRefs);
    assert_eq!(req.service, "test-service");
}

#[test]
fn stream_outcome_bytes_round_trip() {
    assert_eq!(StreamOutcome::Eof.bytes(), None);
    assert_eq!(StreamOutcome::Bytes(7).bytes(), Some(7));
}

#[test]
fn pack_stats_default_is_empty() {
    let s = PackStats::default();
    assert!(s.is_empty());
    assert_eq!(s.bytes_streamed, 0);
    assert_eq!(s.objects_reported, 0);
}

#[test]
fn ingest_report_empty_is_default() {
    let r = IngestReport::empty();
    assert!(r.refs_updated.is_empty());
    assert!(r.stats.is_empty());
}

#[test]
fn empty_readers_and_helpers_are_present() {
    let r1 = empty_reader();
    let r2 = buf_empty_reader();
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("rt");
    let n1 = rt.block_on(async {
        let mut r = r1;
        let mut buf = [0u8; 4];
        r.read(&mut buf).await.unwrap_or(0)
    });
    let n2 = rt.block_on(async {
        let mut r = r2;
        let mut buf = [0u8; 4];
        r.read(&mut buf).await.unwrap_or(0)
    });
    assert_eq!(n1, 0);
    assert_eq!(n2, 0);
}

#[tokio::test]
async fn backend_placeholders_report_unimplemented() {
    let backend = PlaceholderBackend;
    let req = make_request("GET", "/info/refs", EndpointKind::InfoRefs);
    let err = backend
        .authorize(&req)
        .await
        .expect_err("placeholder must return Err");
    match err {
        ServerError::Unimplemented(msg) => assert!(msg.contains("authorize")),
        other => panic!("unexpected error variant: {:?}", other),
    }
}

#[tokio::test]
async fn child_pack_reader_reads_bytes_then_reports_eof() {
    use std::io::Cursor;
    let data: Vec<u8> = (0u8..16).collect();
    let reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data.clone())), None);

    let mut total: u64 = 0;
    let mut reader = reader;
    loop {
        let mut buf = [0u8; 4];
        let n = reader.next_chunk(&mut buf).await.expect("read ok");
        if n == 0 {
            break;
        }
        total = total.saturating_add(n as u64);
    }
    assert_eq!(total, 16);
    assert!(reader.finished());
}

#[tokio::test]
async fn child_pack_reader_collects_pack_stats_and_enforces_cap() {
    use std::io::Cursor;
    let data: Vec<u8> = (0u8..32).collect();
    let mut reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data)), None);
    let stats = reader
        .collect_pack_stats(Some(64))
        .await
        .expect("within cap");
    assert_eq!(stats.bytes_streamed, 32);

    let data: Vec<u8> = (0u8..32).collect();
    let mut reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data)), None);
    let err = reader
        .collect_pack_stats(Some(8))
        .await
        .expect_err("cap should reject");
    match err {
        ServerError::PayloadTooLarge { .. } => {}
        other => panic!("unexpected error: {:?}", other),
    }
}

#[tokio::test]
async fn child_pack_writer_copies_reader_to_writer_and_finishes() {
    use std::io::Cursor;
    use tokio::io::BufWriter;

    let src: Vec<u8> = (0u8..8).collect();
    let mut src_reader = tokio::io::BufReader::new(Cursor::new(src));
    let dst: Vec<u8> = Vec::new();
    let buf_dst = BufWriter::new(Cursor::new(dst));

    let mut writer = ChildPackWriter::new(buf_dst, None);
    let outcome = writer.copy_from(&mut src_reader).await.expect("copy ok");
    match outcome {
        StreamOutcome::Bytes(n) => assert_eq!(n, 8),
        StreamOutcome::Eof => panic!("expected bytes outcome"),
    }
    writer.finish().await.expect("finish ok");
}

#[tokio::test]
async fn server_request_body_streams_bytes() {
    use std::io::Cursor;
    let body_data: Vec<u8> = b"PACK".to_vec();
    let req = ServerRequest::new(
        "POST",
        "/git-receive-pack",
        EndpointKind::GitReceivePack,
        "test",
        Box::new(Cursor::new(body_data.clone())),
    );

    let mut body = req.body;
    let mut got = Vec::new();
    body.read_to_end(&mut got).await.expect("read body");
    assert_eq!(got, body_data);
}

#[test]
fn ref_update_entry_serialises_oid_strings_not_objectid() {
    let entry = RefUpdateEntry {
        name: "refs/heads/x".into(),
        new_oid: "deadbeef".into(),
        old_oid: Some("feedface".into()),
    };
    assert_eq!(entry.new_oid, "deadbeef");
    assert_eq!(entry.old_oid.as_deref(), Some("feedface"));
}

#[test]
fn endpoint_kind_str_matches_module_taxonomy() {
    let _guard = lock_or_recover();
    assert_eq!(EndpointKind::GitUploadPack.as_str(), "git-upload-pack");
    assert_eq!(EndpointKind::GitReceivePack.as_str(), "git-receive-pack");
    assert_eq!(EndpointKind::LfsBatch.as_str(), "lfs-batch");
    assert_eq!(EndpointKind::LfsObjects.as_str(), "lfs-objects");
    assert_eq!(EndpointKind::InfoRefs.as_str(), "info-refs");
    assert_eq!(EndpointKind::SmartHttp.as_str(), "smart-http");
    assert_eq!(EndpointKind::Custom.as_str(), "custom");
}

#[test]
fn guard_clone_is_independent_until_cancel() {
    let g = StreamGuard::new();
    let g2 = g.clone();
    g.cancel();
    assert!(g.is_cancelled());
    assert!(g2.is_cancelled());
}

#[test]
fn auth_context_default_is_anonymous() {
    let ctx = AuthContext::default();
    assert_eq!(ctx.scheme, AuthScheme::Anonymous);
    assert!(ctx.principal.is_none());
}

#[test]
fn server_request_default_output_cap_matches_command_cap() {
    let req = make_request("GET", "/", EndpointKind::Custom);
    assert_eq!(
        req.output_cap,
        gitbaby::server::DEFAULT_OUTPUT_CAP_BYTES as u64
    );
}

#[test]
fn ref_update_batch_default_is_empty() {
    let batch = RefUpdateBatch::default();
    assert!(batch.is_empty());
    assert_eq!(batch.len(), 0);
}

#[test]
fn endpoint_kind_equality_works() {
    assert_eq!(EndpointKind::InfoRefs, EndpointKind::InfoRefs);
    assert_ne!(EndpointKind::InfoRefs, EndpointKind::SmartHttp);
}

#[tokio::test]
async fn stream_guard_cancel_during_async_does_not_panic() {
    let guard = Arc::new(StreamGuard::new());
    let g2 = guard.clone();

    let h = tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(200)).await;
        g2.cancel();
    });

    tokio::time::sleep(Duration::from_millis(50)).await;
    assert!(!guard.is_cancelled());

    h.await.expect("task ok");
    assert!(guard.is_cancelled());
}