use super::*;
use axum::{
    Json, Router,
    http::StatusCode,
    response::IntoResponse,
    routing::{get, post, put},
};
use serde_json::json;
use std::{
    collections::BTreeMap,
    sync::atomic::{AtomicUsize, Ordering},
};
use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    sync::{oneshot, watch},
    task::JoinHandle,
};

struct TestServer {
    base: String,
    stop: Option<oneshot::Sender<()>>,
    task: JoinHandle<()>,
    _directory: Option<tempfile::TempDir>,
}

impl TestServer {
    async fn fixture(router: Router) -> Self {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://{}", listener.local_addr().unwrap());
        let (stop, done) = oneshot::channel();
        let task = tokio::spawn(async move {
            axum::serve(listener, router)
                .with_graceful_shutdown(async {
                    let _ = done.await;
                })
                .await
                .unwrap();
        });
        Self {
            base,
            stop: Some(stop),
            task,
            _directory: None,
        }
    }

    async fn production(token: Option<String>, prefix: bool) -> Self {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().to_owned();
        let store = Arc::new(
            tokio::task::spawn_blocking(move || oa_server::store::Store::open(path))
                .await
                .unwrap()
                .unwrap(),
        );
        let router = oa_server::app(store, token);
        let router = if prefix {
            Router::new().nest("/teams/engineering", router)
        } else {
            router
        };
        let mut server = Self::fixture(router).await;
        server._directory = Some(directory);
        if prefix {
            server.base.push_str("/teams/engineering");
        }
        server
    }

    fn connection(&self) -> Connection {
        Connection::new(&self.base, Secret::default()).unwrap()
    }

    async fn finish(mut self) {
        self.stop.take().unwrap().send(()).unwrap();
        (&mut self.task).await.unwrap();
    }
}

impl Drop for TestServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

fn authored() -> SourceBundle {
    let bundle = SourceBundle {
        entry: "harbor.arcadia".into(),
        files: BTreeMap::from([
            ("harbor.arcadia".into(), "// Harbor rationale 航海\r\narcadia 4; project harbor \"Harbor & rescue\" {\r\n include \"work/operations.arcadia\";\r\n}\r\n".into()),
            ("work/operations.arcadia".into(), "// Keep this authored organization\r\noa_activity assess \"Assess assistance\" {}\r\nsa_function assist \"Assist operator\" {}\r\n".into()),
        ]),
    };
    source_project::compile_bundle(&bundle, SOURCE_LIMITS)
        .unwrap()
        .bundle
}

fn envelope(bundle: SourceBundle, resource_id: Id, revision: u64) -> serde_json::Value {
    let model = source_project::compile_bundle(&bundle, SOURCE_LIMITS)
        .unwrap()
        .decoded
        .project;
    json!({"resource_id":resource_id,"revision":revision,"model":model,"source_project":bundle})
}

#[test]
fn hosted_endpoint_validation_redacts_secrets_and_preserves_proxy_paths() {
    for base in [
        "http://example.com",
        "http://192.168.1.4",
        "ftp://localhost",
        "not a url",
        "https://user:secret@example.com",
        "https://@example.com",
        "https://user@example.com",
        "https://example.com/?token=secret",
        "https://example.com/#secret",
        "https://example.com\\hidden",
        "https://example.com/\nsecret",
    ] {
        let failure = Connection::new(base, Secret("secret".into())).unwrap_err();
        assert_eq!(failure.code, "invalid_endpoint");
        assert!(!failure.detail.contains("secret"));
        assert!(!failure.uncertain);
    }
    for base in [
        "http://localhost:8787",
        "http://127.0.0.2:8787",
        "http://[::1]:8787",
        "https://example.com/prefix",
    ] {
        assert!(Connection::new(base, Secret::default()).is_ok(), "{base}");
    }
    let client = Connection::new(
        "https://example.com/teams/space%20name",
        Secret("private-token".into()),
    )
    .unwrap();
    assert_eq!(client.base(), "https://example.com/teams/space%20name/");
    assert_eq!(
        client.endpoint("projects").as_str(),
        "https://example.com/teams/space%20name/api/v1/projects"
    );
    assert!(!format!("{client:?}").contains("private-token"));
    assert_eq!(
        format!("{:?}", Secret("private-token".into())),
        "Secret([redacted])"
    );
    assert_eq!(
        Connection::new("http://localhost", Secret("bad\nheader".into()))
            .unwrap_err()
            .code,
        "invalid_token"
    );
    for etag in [
        "*",
        "W/\"weak\"",
        "unquoted",
        "\"one\",\"two\"",
        "\"new\nline\"",
    ] {
        assert!(!valid_etag(etag), "{etag}");
    }
    assert!(valid_etag("\"opaque:tag/+,=?\""));
}

#[tokio::test]
async fn hosted_production_roundtrip_preserves_comments_noops_and_conflicts() {
    let server = TestServer::production(None, false).await;
    let client = server.connection();
    assert!(client.catalog("", None).await.unwrap().items.is_empty());
    let bundle = authored();
    let original = client.publish(bundle.clone()).await.unwrap();
    assert_ne!(original.resource_id, original.compiled.decoded.project.id);
    assert_eq!(original.compiled.bundle, bundle);
    assert_eq!(original.revision, 1);
    let catalog = client.catalog("Harbor &", None).await.unwrap();
    assert_eq!(catalog.items.len(), 1);
    assert_eq!(catalog.items[0].resource_id, original.resource_id);
    assert_eq!(
        catalog.items[0].model_id,
        original.compiled.decoded.project.id
    );
    assert_eq!(catalog.items[0].name, "Harbor & rescue");
    assert_eq!(
        (catalog.items[0].elements, catalog.items[0].relations),
        (2, 0)
    );
    assert!(
        client
            .catalog("?q=missing&limit=1000", None)
            .await
            .unwrap()
            .items
            .is_empty()
    );
    assert!(
        client
            .catalog("", Some(original.resource_id))
            .await
            .unwrap()
            .items
            .is_empty()
    );
    let same = client
        .save(original.resource_id, &original.etag, bundle.clone())
        .await
        .unwrap();
    assert_eq!(same.etag, original.etag);
    assert_eq!(same.revision, original.revision);
    let mut changed = bundle.clone();
    changed
        .files
        .get_mut("work/operations.arcadia")
        .unwrap()
        .push_str("// Reviewed by the team\r\n");
    let saved = client
        .save(original.resource_id, &original.etag, changed.clone())
        .await
        .unwrap();
    assert_eq!(saved.resource_id, original.resource_id);
    assert_eq!(saved.revision, 2);
    assert_ne!(saved.etag, original.etag);
    assert_eq!(saved.compiled.bundle, changed);
    assert_eq!(
        saved.compiled.decoded.project,
        original.compiled.decoded.project
    );
    let conflict = client
        .save(original.resource_id, &original.etag, bundle)
        .await
        .unwrap_err();
    assert_eq!(conflict.status, Some(412));
    assert_eq!(conflict.code, "revision_conflict");
    assert!(!conflict.uncertain);
    assert_eq!(
        client
            .open(original.resource_id)
            .await
            .unwrap()
            .compiled
            .bundle,
        changed
    );
    server.finish().await;
}

#[tokio::test]
async fn hosted_production_auth_proxy_prefix_and_located_rejection() {
    let token = "hosted-test-token-with-at-least-32-characters";
    let server = TestServer::production(Some(token.into()), true).await;
    let unauthorized = server.connection().catalog("", None).await.unwrap_err();
    assert_eq!(
        (unauthorized.status, unauthorized.code.as_str()),
        (Some(401), "unauthorized")
    );
    let client = Connection::new(&server.base, Secret(token.into())).unwrap();
    let original = client.publish(authored()).await.unwrap();
    assert_eq!(
        client.catalog("Harbor &", None).await.unwrap().items.len(),
        1
    );
    let mut invalid = original.compiled.bundle.clone();
    invalid
        .files
        .get_mut("work/operations.arcadia")
        .unwrap()
        .push_str("this is not valid source;");
    let rejected = client
        .save(original.resource_id, &original.etag, invalid)
        .await
        .unwrap_err();
    assert_eq!(rejected.status, Some(422));
    assert!(!rejected.uncertain);
    assert!(rejected.detail.contains("work/operations.arcadia:"));
    assert_eq!(
        client.open(original.resource_id).await.unwrap().etag,
        original.etag
    );
    let missing = client.open(Id::new_v4()).await.unwrap_err();
    assert_eq!(missing.status, Some(404));
    assert!(!missing.uncertain);
    server.finish().await;
}

#[tokio::test]
async fn hosted_redirect_is_not_followed_or_retried_and_problem_tokens_are_redacted() {
    let redirected = Arc::new(AtomicUsize::new(0));
    let calls = Arc::clone(&redirected);
    let destination = TestServer::fixture(Router::new().fallback(move || {
        let calls = Arc::clone(&calls);
        async move {
            calls.fetch_add(1, Ordering::SeqCst);
            StatusCode::OK
        }
    }))
    .await;
    let location = destination.base.clone();
    let attempts = Arc::new(AtomicUsize::new(0));
    let seen = Arc::clone(&attempts);
    let server = TestServer::fixture(Router::new().route("/api/v1/projects/import-source-project", post(move || {
        let location = location.clone();
        let seen = Arc::clone(&seen);
        async move { seen.fetch_add(1, Ordering::SeqCst); (StatusCode::TEMPORARY_REDIRECT, [(header::LOCATION, location)]) }
    })).route("/api/v1/projects", get(|| async {
        (StatusCode::FORBIDDEN, Json(json!({"code":"denied","detail":"Do not print private-test-token", "issues":[{"message":"private-test-token","path":"secret/private-test-token.arcadia","line":4,"column":2}]})))
    }))).await;
    let client = Connection::new(&server.base, Secret("private-test-token".into())).unwrap();
    let failure = client.publish(authored()).await.unwrap_err();
    assert_eq!(failure.code, "redirect_refused");
    assert!(!failure.uncertain);
    assert_eq!(attempts.load(Ordering::SeqCst), 1);
    assert_eq!(redirected.load(Ordering::SeqCst), 0);
    let failure = client.catalog("", None).await.unwrap_err();
    assert!(failure.detail.contains("[redacted]"));
    assert!(!format!("{failure:?}").contains("private-test-token"));
    server.finish().await;
    destination.finish().await;
}

#[tokio::test]
async fn hosted_open_rejects_incoherent_envelopes_missing_etags_and_unstable_ids() {
    let id = Id::new_v4();
    let stable = authored();
    let mut mismatched = envelope(stable.clone(), id, 1);
    mismatched["model"]["name"] = json!("different model");
    let mut wrong_id = envelope(stable.clone(), Id::new_v4(), 1);
    // Keep this independently useful even if the model's schema changes.
    wrong_id["revision"] = json!(1);
    let mut unstable = envelope(stable.clone(), id, 1);
    unstable["source_project"] = json!({"entry":"model.arcadia","files":{"model.arcadia":"arcadia 4; project unstable \"Unstable\" {}"}});
    for (body, etag) in [
        (mismatched, Some("\"opaque\"")),
        (wrong_id, Some("\"opaque\"")),
        (envelope(stable.clone(), id, 0), Some("\"opaque\"")),
        (envelope(stable, id, 1), None),
        (unstable, Some("\"opaque\"")),
    ] {
        let server = TestServer::fixture(Router::new().route(
            "/api/v1/projects/{id}",
            get(move || {
                let body = body.clone();
                async move {
                    let mut response = Json(body).into_response();
                    if let Some(etag) = etag {
                        response
                            .headers_mut()
                            .insert(header::ETAG, etag.parse().unwrap());
                    }
                    response
                }
            }),
        ))
        .await;
        let failure = server.connection().open(id).await.unwrap_err();
        assert_eq!(failure.code, "invalid_response");
        assert!(!failure.uncertain);
        server.finish().await;
    }
}

#[tokio::test]
async fn hosted_put_preserves_an_opaque_etag_and_compiles_the_accepted_bundle() {
    let id = Id::new_v4();
    let etag = "\"opaque:tag/+,=?\"";
    let bundle = authored();
    let response_bundle = bundle.clone();
    let server = TestServer::fixture(Router::new().route(
        "/api/v1/projects/{id}/source-project",
        put(
            move |headers: header::HeaderMap, Json(body): Json<SourceBundle>| {
                let expected_bundle = response_bundle.clone();
                async move {
                    assert_eq!(headers.get(header::IF_MATCH).unwrap(), etag);
                    assert_eq!(body, expected_bundle);
                    (
                        [(header::ETAG, "\"accepted:new-tag\"")],
                        Json(json!({"revision":8,"changed":false,"source_project":body})),
                    )
                }
            },
        ),
    ))
    .await;
    let client = server.connection();
    let saved = client.save(id, etag, bundle.clone()).await.unwrap();
    assert_eq!(saved.etag, "\"accepted:new-tag\"");
    assert_eq!(saved.resource_id, id);
    assert_eq!(saved.revision, 8);
    assert_eq!(saved.compiled.bundle, bundle.clone());
    assert_eq!(
        client.save(id, "*", bundle).await.unwrap_err().code,
        "invalid_etag"
    );
    server.finish().await;
}

#[tokio::test]
async fn hosted_publication_retains_known_resource_after_failed_followup() {
    let id = Id::new_v4();
    let attempts = Arc::new(AtomicUsize::new(0));
    let calls = Arc::clone(&attempts);
    let server = TestServer::fixture(Router::new().route("/api/v1/projects/import-source-project", post(move || {
        let calls = Arc::clone(&calls);
        async move {
            calls.fetch_add(1, Ordering::SeqCst);
            (StatusCode::CREATED, [(header::ETAG, "\"receipt\"")], Json(json!({"resource_id":id,"revision":1,"model_id":Id::new_v4(),"name":"Published","elements":0,"relations":0})))
        }
    })).route("/api/v1/projects/{id}", get(|| async { StatusCode::SERVICE_UNAVAILABLE }))).await;
    let failure = server.connection().publish(authored()).await.unwrap_err();
    assert_eq!(failure.published, Some(id));
    assert!(failure.uncertain);
    assert_eq!(failure.status, Some(503));
    assert_eq!(attempts.load(Ordering::SeqCst), 1);
    server.finish().await;
}

/// Reads the entire request before simulating a lost/malformed response. The
/// counter proves the production client never repeats an admitted publication.
async fn raw_fixture(reply: Vec<u8>) -> (String, Arc<AtomicUsize>, JoinHandle<()>) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let base = format!("http://{}", listener.local_addr().unwrap());
    let attempts = Arc::new(AtomicUsize::new(0));
    let calls = Arc::clone(&attempts);
    let task = tokio::spawn(async move {
        loop {
            let (mut stream, _) = listener.accept().await.unwrap();
            let mut request = Vec::new();
            let mut buffer = [0; 4096];
            loop {
                let read = stream.read(&mut buffer).await.unwrap();
                if read == 0 {
                    break;
                }
                request.extend_from_slice(&buffer[..read]);
                if let Some(end) = request.windows(4).position(|part| part == b"\r\n\r\n") {
                    let headers = String::from_utf8_lossy(&request[..end]).to_ascii_lowercase();
                    let length = headers
                        .lines()
                        .find_map(|line| {
                            line.strip_prefix("content-length:")
                                .and_then(|value| value.trim().parse::<usize>().ok())
                        })
                        .unwrap_or(0);
                    if request.len() >= end + 4 + length {
                        break;
                    }
                }
            }
            calls.fetch_add(1, Ordering::SeqCst);
            let _ = stream.write_all(&reply).await;
            let _ = stream.shutdown().await;
        }
    });
    (base, attempts, task)
}

#[tokio::test]
async fn hosted_lost_publication_response_is_uncertain_without_retry() {
    let (base, attempts, task) = raw_fixture(Vec::new()).await;
    let client = Connection::new(&base, Secret::default()).unwrap();
    let failure = client.publish(authored()).await.unwrap_err();
    assert!(failure.uncertain);
    assert_eq!(failure.published, None);
    assert_eq!(attempts.load(Ordering::SeqCst), 1);
    task.abort();
}

#[tokio::test]
async fn hosted_response_limits_cover_declared_and_chunked_bodies() {
    let declared = format!(
        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
        MAX_RESPONSE_BYTES + 1
    )
    .into_bytes();
    let mut chunked =
        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n".to_vec();
    let chunk = vec![b' '; 64 * 1024];
    for _ in 0..=(MAX_RESPONSE_BYTES / chunk.len()) {
        chunked.extend_from_slice(format!("{:x}\r\n", chunk.len()).as_bytes());
        chunked.extend_from_slice(&chunk);
        chunked.extend_from_slice(b"\r\n");
    }
    chunked.extend_from_slice(b"0\r\n\r\n");
    for reply in [declared, chunked] {
        let (base, _, task) = raw_fixture(reply).await;
        let failure = Connection::new(&base, Secret::default())
            .unwrap()
            .catalog("", None)
            .await
            .unwrap_err();
        assert_eq!(failure.code, "response_too_large");
        assert!(!failure.uncertain);
        task.abort();
    }
}

#[tokio::test]
async fn hosted_timeout_and_disconnect_leave_mutation_outcomes_explicit() {
    let received = Arc::new(AtomicUsize::new(0));
    let calls = Arc::clone(&received);
    let (release, held) = watch::channel(false);
    let server = TestServer::fixture(Router::new().route(
        "/api/v1/projects/import-source-project",
        post(move || {
            let calls = Arc::clone(&calls);
            let mut held = held.clone();
            async move {
                calls.fetch_add(1, Ordering::SeqCst);
                // Never race a fixed response delay against the client deadline.
                // A retained watch value also releases a handler admitted late.
                while !*held.borrow_and_update() {
                    held.changed().await.unwrap();
                }
                StatusCode::CREATED
            }
        }),
    ))
    .await;
    let client =
        Connection::with_timeout(&server.base, Secret::default(), Duration::from_millis(30))
            .unwrap();
    let failure = client.publish(authored()).await.unwrap_err();
    assert_eq!(failure.code, "timeout");
    assert!(failure.uncertain);
    release.send(true).unwrap();
    let base = server.base.clone();
    server.finish().await;
    // Under executor/CPU contention, the deadline can expire before admission.
    // Both zero and one admission have an uncertain mutation outcome; neither
    // permits a retry. The lost-response test above independently reads a full
    // admitted request and requires exactly one publication.
    assert!(received.load(Ordering::SeqCst) <= 1);
    let failure = Connection::with_timeout(&base, Secret::default(), Duration::from_secs(2))
        .unwrap()
        .catalog("", None)
        .await
        .unwrap_err();
    assert!(matches!(
        failure.code.as_str(),
        "connection_failed" | "timeout"
    ));
    assert!(!failure.uncertain);
}

#[tokio::test]
async fn hosted_publication_race_keeps_receipt_identity_for_review() {
    let id = Id::new_v4();
    let bundle = authored();
    let published_model_id = source_project::compile_bundle(&bundle, SOURCE_LIMITS)
        .unwrap()
        .decoded
        .project
        .id;
    let mut changed = bundle.clone();
    changed
        .files
        .get_mut("work/operations.arcadia")
        .unwrap()
        .push_str("// Concurrent external review\r\n");
    let head = envelope(changed, id, 2);
    let server = TestServer::fixture(Router::new().route("/api/v1/projects/import-source-project", post(move || async move {
        (StatusCode::CREATED, [(header::ETAG, "\"first-revision\"")], Json(json!({"resource_id":id,"revision":1,"model_id":published_model_id,"name":"Published","elements":2,"relations":0})))
    })).route("/api/v1/projects/{id}", get(move || {
        let head = head.clone();
        async move { ([(header::ETAG, "\"next-revision\"")], Json(head)) }
    }))).await;
    let failure = server.connection().publish(bundle).await.unwrap_err();
    assert!(failure.uncertain);
    assert_eq!(failure.published, Some(id));
    assert!(failure.detail.contains("changed before"));
    server.finish().await;
}

#[tokio::test]
async fn hosted_structured_mutation_failures_distinguish_precommit_rejections() {
    for (status, code, uncertain) in [
        (StatusCode::PAYLOAD_TOO_LARGE, "request_too_large", false),
        (StatusCode::INSUFFICIENT_STORAGE, "store_capacity", false),
        (StatusCode::SERVICE_UNAVAILABLE, "overloaded", false),
        (StatusCode::SERVICE_UNAVAILABLE, "storage_unavailable", true),
        (StatusCode::GATEWAY_TIMEOUT, "outcome_unknown", true),
    ] {
        let server = TestServer::fixture(Router::new().route(
            "/api/v1/projects/import-source-project",
            post(move || async move {
                (
                    status,
                    Json(json!({"code":code,"detail":"Service-specific actionable detail"})),
                )
            }),
        ))
        .await;
        let failure = server.connection().publish(authored()).await.unwrap_err();
        assert_eq!(failure.status, Some(status.as_u16()));
        assert_eq!(failure.code, code);
        assert_eq!(failure.uncertain, uncertain, "{code}");
        assert_eq!(failure.detail, "Service-specific actionable detail");
        server.finish().await;
    }
}

#[tokio::test]
async fn hosted_oversized_local_bundle_is_rejected_before_network_admission() {
    let mut bundle = authored();
    bundle
        .files
        .get_mut(&bundle.entry)
        .unwrap()
        .push_str(&" ".repeat(8 * 1024 * 1024));
    let failure = Connection::new("http://127.0.0.1:1", Secret::default())
        .unwrap()
        .publish(bundle)
        .await
        .unwrap_err();
    assert_eq!(failure.code, "request_too_large");
    assert!(!failure.uncertain);
    assert!(failure.published.is_none());
}