// Copyright 2026 PingCAP, Inc.

use builtin_dfs::{BuiltinDfs, BuiltinDfsFileMeta};
use bytes::Bytes;
use http::{Request, StatusCode};
use hyper::{Body, Client, client::HttpConnector};
use kvengine::dfs::{Dfs, FileType, Options};
use pd_client::PdClient;
use test_cloud_server::{ServerCluster, ServerClusterBuilder, alloc_node_id_vec};
use tikv::config::TikvConfig;
use tikv_util::{info, sys::SysQuota};
use tokio::runtime::Runtime;

/// Test context for BuiltinDfs integration tests.
struct BuiltinDfsTestContext {
    cluster: ServerCluster,
    builtin_dfs: BuiltinDfs,
    status_addr: String,
    http_client: Client<HttpConnector>,
    runtime: Runtime,
    opts: Options,
}

impl BuiltinDfsTestContext {
    /// Create a new test context with a cluster and BuiltinDfs client.
    fn new() -> Self {
        Self::new_with_read_byte_limit(None)
    }

    fn new_with_read_byte_limit(max_in_flight_bytes: Option<u64>) -> Self {
        // Start a cluster with builtin DFS
        let nodes = alloc_node_id_vec(1);
        let cluster = ServerClusterBuilder::new(nodes.clone(), move |_, conf: &mut TikvConfig| {
            conf.dfs.s3_endpoint = "local".to_string();
            if let Some(max_in_flight_bytes) = max_in_flight_bytes {
                conf.kvengine.dfs_load_memory_ratio =
                    max_in_flight_bytes as f64 / SysQuota::memory_limit_in_bytes().max(1) as f64;
            }
        })
        .pd_server_cnt(1)
        .build();

        cluster.wait_region_replicated(&[], 1);

        let node_id = nodes[0];
        let status_addr = cluster.status_addr(node_id);
        let http_client = Client::new();
        let runtime = Runtime::new().unwrap();

        // Create BuiltinDfs using the cluster's PD client
        let pd_client = cluster.get_pd_client();
        let builtin_dfs = BuiltinDfs::new(pd_client.clone());

        // Get actual region info
        let region = pd_client.get_region(b"").unwrap();
        let region_id = region.get_id();
        let region_ver = region.get_region_epoch().get_version();
        let opts = Options::default().with_shard(region_id, region_ver);

        Self {
            cluster,
            builtin_dfs,
            status_addr,
            http_client,
            runtime,
            opts,
        }
    }

    /// Create a file via BuiltinDfs and wait for it to be written.
    fn create_file(&self, file_id: u64, file_data: &[u8]) {
        self.create_file_with_opts(file_id, file_data, self.opts);
    }

    fn create_file_with_opts(&self, file_id: u64, file_data: &[u8], opts: Options) {
        info!("Creating file {} with {} bytes", file_id, file_data.len());
        self.runtime
            .block_on(
                self.builtin_dfs
                    .create(file_id, Bytes::from(file_data.to_vec()), opts),
            )
            .expect("Failed to create file via BuiltinDfs");

        // Wait a bit for file to be written
        std::thread::sleep(std::time::Duration::from_millis(100));
    }

    /// Send an HTTP GET request and return the response.
    fn http_get(&self, url: &str) -> hyper::Response<Body> {
        let req = Request::get(url).body(Body::empty()).unwrap();
        self.runtime
            .block_on(self.http_client.request(req))
            .unwrap()
    }

    /// Stop the cluster.
    fn stop(mut self) {
        self.cluster.stop();
    }
}

/// Test status_server meta API via BuiltinDfs and HTTP.
///
/// This test validates the complete meta API functionality:
/// 1. HTTP API returns 404 for non-existent files
/// 2. HTTP API returns 200 with JSON metadata for existing files
/// 3. BuiltinDfs::size() correctly parses file size from response
/// 4. BuiltinDfs::exists() correctly determines file existence
#[test]
fn test_builtin_dfs_meta() {
    let ctx = BuiltinDfsTestContext::new();

    // Test parameters
    let file_id = 12345u64;
    let file_type = FileType::Sst;
    let file_data = b"Hello, status_server meta API test!";
    let expected_size = file_data.len() as u64;

    // HTTP endpoint for meta query
    let meta_url = format!(
        "http://{}/dfs/{}?file_type={}&meta=true",
        ctx.status_addr,
        file_id,
        file_type.suffix()
    );

    info!("Test 1: meta query before file creation - should return 404");
    let resp = ctx.http_get(&meta_url);
    assert_eq!(
        resp.status(),
        StatusCode::NOT_FOUND,
        "Should return 404 for non-existent file"
    );

    info!("Test 2: BuiltinDfs::exists before file creation - should return false");
    let exists = ctx
        .runtime
        .block_on(ctx.builtin_dfs.exists(file_id, ctx.opts))
        .expect("BuiltinDfs::exists should succeed");
    assert!(!exists, "File should not exist");

    // Create the file
    ctx.create_file(file_id, file_data);

    info!("Test 3: meta query after file creation - should return 200 with metadata");
    let resp = ctx.http_get(&meta_url);
    assert_eq!(
        resp.status(),
        StatusCode::OK,
        "Should return 200 for existing file"
    );

    // Parse JSON response body
    let body_bytes = ctx
        .runtime
        .block_on(hyper::body::to_bytes(resp.into_body()))
        .unwrap();

    let info: BuiltinDfsFileMeta =
        serde_json::from_slice(&body_bytes).expect("Failed to parse meta JSON response");
    assert_eq!(
        info.size, expected_size,
        "File size should match: expected {}, got {}",
        expected_size, info.size
    );

    info!("Test 4: BuiltinDfs::size - should return correct file size");
    let size = ctx
        .runtime
        .block_on(ctx.builtin_dfs.size(file_id, ctx.opts))
        .expect("BuiltinDfs::size should succeed");
    assert_eq!(size, expected_size, "File size should match");

    info!("Test 5: BuiltinDfs::exists after file creation - should return true");
    let exists = ctx
        .runtime
        .block_on(ctx.builtin_dfs.exists(file_id, ctx.opts))
        .expect("BuiltinDfs::exists should succeed");
    assert!(exists, "File should exist");

    info!("All tests completed successfully!");
    ctx.stop();
}

#[test]
fn test_builtin_dfs_streamed_full_and_range_read() {
    let ctx = BuiltinDfsTestContext::new();
    let file_id = 12346u64;
    let file_data = (0..2 * 1024 * 1024 + 37)
        .map(|i| (i % u8::MAX as usize) as u8)
        .collect::<Vec<_>>();

    // Create a file large enough to span multiple response chunks.
    ctx.create_file(file_id, &file_data);

    // Verify the existing BuiltinDfs client can consume the full streamed body.
    let full_data = ctx
        .runtime
        .block_on(ctx.builtin_dfs.read_file(file_id, ctx.opts))
        .expect("Failed to read full file via BuiltinDfs");
    assert_eq!(full_data.as_ref(), file_data.as_slice());

    // Verify a range crossing a response chunk boundary is unchanged.
    let start_off = 123usize;
    let end_off = 1024 * 1024 + 456;
    let range_opts = ctx
        .opts
        .with_start_off(start_off as u64)
        .with_end_off(end_off as u64);
    let range_data = ctx
        .runtime
        .block_on(ctx.builtin_dfs.read_file(file_id, range_opts))
        .expect("Failed to read file range via BuiltinDfs");
    assert_eq!(range_data.as_ref(), &file_data[start_off..end_off]);

    // Explicit ranges are admitted from their requested length without a
    // separate size lookup, while a range past EOF remains a bad request.
    let invalid_range_url = format!(
        "http://{}/dfs/{}?file_type={}&start_off=0&end_off={}",
        ctx.status_addr,
        file_id,
        FileType::Sst.suffix(),
        file_data.len() + 1,
    );
    let response = ctx.http_get(&invalid_range_url);
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    let empty_range_past_eof_url = format!(
        "http://{}/dfs/{}?file_type={}&start_off={}&end_off={}",
        ctx.status_addr,
        file_id,
        FileType::Sst.suffix(),
        file_data.len() + 1,
        file_data.len() + 1,
    );
    let response = ctx.http_get(&empty_range_past_eof_url);
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    ctx.stop();
}

#[test]
fn test_builtin_dfs_byte_admission_rejects_oversized_and_allows_range() {
    let byte_limit = 1024 * 1024;
    let ctx = BuiltinDfsTestContext::new_with_read_byte_limit(Some(byte_limit as u64));
    let file_id = 12347u64;
    let file_data = vec![42; byte_limit * 2];
    ctx.create_file(file_id, &file_data);

    // A full read larger than the complete byte budget is rejected before the
    // source buffer is allocated.
    let full_read_url = format!(
        "http://{}/dfs/{}?file_type={}",
        ctx.status_addr,
        file_id,
        FileType::Sst.suffix()
    );
    let response = ctx.http_get(&full_read_url);
    assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);

    // A range from the same file is charged by its range length and succeeds.
    let range_end = (byte_limit / 2) as u64;
    let range_opts = ctx.opts.with_end_off(range_end);
    let range_data = ctx
        .runtime
        .block_on(ctx.builtin_dfs.read_file(file_id, range_opts))
        .expect("Failed to read admitted file range via BuiltinDfs");
    assert_eq!(range_data.as_ref(), &file_data[..byte_limit / 2]);

    // Txn chunk range parameters are ignored by the read path, so even a tiny
    // requested range must be charged by the complete chunk size.
    let txn_chunk_id = 12348u64;
    let txn_opts = ctx.opts.with_type(FileType::TxnChunk);
    ctx.create_file_with_opts(txn_chunk_id, &file_data, txn_opts);
    let txn_range_url = format!(
        "http://{}/dfs/{}?file_type={}&start_off=0&end_off=1",
        ctx.status_addr,
        txn_chunk_id,
        FileType::TxnChunk.suffix()
    );
    let response = ctx.http_get(&txn_range_url);
    assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);

    // Range validation happens before admission, so an out-of-file range
    // remains a bad request even when its length exceeds the byte budget.
    let invalid_range_url = format!(
        "http://{}/dfs/{}?file_type={}&start_off=0&end_off={}",
        ctx.status_addr,
        file_id,
        FileType::Sst.suffix(),
        file_data.len() + byte_limit,
    );
    let response = ctx.http_get(&invalid_range_url);
    assert_eq!(response.status(), StatusCode::BAD_REQUEST);

    ctx.stop();
}