use metrics::{counter, gauge, histogram};
use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};

/// Initialize the global metrics recorder and return a handle for the
/// Prometheus scrape endpoint.
///
/// Called once during server startup. Subsequent calls to `counter!`,
/// `gauge!`, and `histogram!` anywhere in the codebase will record into
/// this registry.
pub fn init_metrics() -> PrometheusHandle {
    PrometheusBuilder::new()
        .install_recorder()
        .expect("failed to install Prometheus metrics recorder")
}

// ── Metric names (constants prevent typos across crates) ──

/// Total HTTP requests received, labeled by method, path pattern, and status.
pub const HTTP_REQUESTS_TOTAL: &str = "conman_http_requests_total";

/// HTTP request duration in seconds, labeled by method and path pattern.
pub const HTTP_REQUEST_DURATION_SECONDS: &str = "conman_http_request_duration_seconds";

/// Total jobs enqueued, labeled by job type.
pub const JOBS_ENQUEUED_TOTAL: &str = "conman_jobs_enqueued_total";

/// Total jobs completed, labeled by job type and outcome (succeeded/failed).
pub const JOBS_COMPLETED_TOTAL: &str = "conman_jobs_completed_total";

/// Job processing duration in seconds, labeled by job type.
pub const JOB_DURATION_SECONDS: &str = "conman_job_duration_seconds";

/// Current number of jobs in `queued` state, labeled by job type.
pub const JOB_QUEUE_DEPTH: &str = "conman_job_queue_depth";

/// Total deployments attempted, labeled by outcome (succeeded/failed/canceled).
pub const DEPLOYMENTS_TOTAL: &str = "conman_deployments_total";

/// Total gitaly gRPC calls, labeled by method and outcome.
pub const GITALY_CALLS_TOTAL: &str = "conman_gitaly_calls_total";

/// Gitaly gRPC call duration in seconds, labeled by method.
pub const GITALY_CALL_DURATION_SECONDS: &str = "conman_gitaly_call_duration_seconds";

/// Total authentication failures (bad password, expired token, etc.).
pub const AUTH_FAILURES_TOTAL: &str = "conman_auth_failures_total";

/// Total rate-limited requests.
pub const RATE_LIMITED_TOTAL: &str = "conman_rate_limited_total";

use axum::{extract::Request, middleware::Next, response::Response};
use metrics::{counter, histogram};
use std::time::Instant;

use crate::metrics::{HTTP_REQUESTS_TOTAL, HTTP_REQUEST_DURATION_SECONDS};

/// Middleware that records HTTP request count and latency.
///
/// Labels: `method`, `path` (matched route pattern, not raw URL), `status`.
/// Must be applied as an outer layer so it captures the full request lifecycle.
pub async fn http_metrics_middleware(req: Request, next: Next) -> Response {
    let method = req.method().to_string();

    // Use the matched path pattern (e.g. "/api/apps/:appId") to avoid
    // high-cardinality labels from path parameters.
    let path = req
        .extensions()
        .get::<axum::extract::MatchedPath>()
        .map(|mp| mp.as_str().to_string())
        .unwrap_or_else(|| "unknown".to_string());

    let start = Instant::now();
    let response = next.run(req).await;
    let duration = start.elapsed().as_secs_f64();

    let status = response.status().as_u16().to_string();

    counter!(HTTP_REQUESTS_TOTAL, "method" => method.clone(), "path" => path.clone(), "status" => status);
    histogram!(HTTP_REQUEST_DURATION_SECONDS, "method" => method, "path" => path).record(duration);

    response
}

use axum::extract::State;
use axum::http::StatusCode;
use axum::Json;
use serde::Serialize;

use crate::state::AppState;

/// Component-level health status.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ComponentStatus {
    /// Component is reachable and responding within acceptable latency.
    Healthy,
    /// Component is reachable but degraded (e.g. high latency, replica lag).
    Degraded,
    /// Component is unreachable or returning errors.
    Unhealthy,
}

/// Individual component health report.
#[derive(Debug, Clone, Serialize)]
pub struct ComponentHealth {
    pub name: &'static str,
    pub status: ComponentStatus,
    /// Human-readable detail (e.g. "ping: 2ms", "connection refused").
    #[serde(skip_serializing_if = "Option::is_none")]
    pub detail: Option<String>,
}

/// Enhanced health response with component-level breakdown.
#[derive(Debug, Clone, Serialize)]
pub struct HealthResponse {
    /// Overall status: "ok" if all components healthy, "degraded" otherwise.
    pub status: &'static str,
    /// Application version from compile-time env.
    pub version: &'static str,
    /// Individual component checks.
    pub components: Vec<ComponentHealth>,
}

/// GET /api/health
///
/// Returns detailed health status for each dependency. Returns 200 when all
/// components are healthy, 503 when any component is unhealthy. Does not
/// require authentication.
pub async fn health_check(State(state): State<AppState>) -> (StatusCode, Json<HealthResponse>) {
    let mut components = Vec::with_capacity(3);
    let mut all_healthy = true;

    // Check MongoDB connectivity.
    let mongo_health = match check_mongo(&state).await {
        Ok(detail) => ComponentHealth {
            name: "mongodb",
            status: ComponentStatus::Healthy,
            detail: Some(detail),
        },
        Err(detail) => {
            all_healthy = false;
            ComponentHealth {
                name: "mongodb",
                status: ComponentStatus::Unhealthy,
                detail: Some(detail),
            }
        }
    };
    components.push(mongo_health);

    // Check Gitaly gRPC channel.
    let gitaly_health = match check_gitaly(&state).await {
        Ok(detail) => ComponentHealth {
            name: "gitaly",
            status: ComponentStatus::Healthy,
            detail: Some(detail),
        },
        Err(detail) => {
            all_healthy = false;
            ComponentHealth {
                name: "gitaly",
                status: ComponentStatus::Unhealthy,
                detail: Some(detail),
            }
        }
    };
    components.push(gitaly_health);

    // Check job runner liveness.
    let job_runner_health = match check_job_runner(&state).await {
        Ok(detail) => ComponentHealth {
            name: "job_runner",
            status: ComponentStatus::Healthy,
            detail: Some(detail),
        },
        Err(detail) => {
            all_healthy = false;
            ComponentHealth {
                name: "job_runner",
                status: ComponentStatus::Unhealthy,
                detail: Some(detail),
            }
        }
    };
    components.push(job_runner_health);

    let status = if all_healthy { "ok" } else { "degraded" };
    let http_status = if all_healthy {
        StatusCode::OK
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    };

    (
        http_status,
        Json(HealthResponse {
            status,
            version: env!("CARGO_PKG_VERSION"),
            components,
        }),
    )
}

/// Ping MongoDB and return round-trip time.
async fn check_mongo(state: &AppState) -> Result<String, String> {
    let start = std::time::Instant::now();
    conman_db::check_mongo_health(&state.db)
        .await
        .map(|_| format!("ping: {}ms", start.elapsed().as_millis()))
        .map_err(|e| e.to_string())
}

/// Verify the Gitaly gRPC channel is connected.
async fn check_gitaly(state: &AppState) -> Result<String, String> {
    match &state.gitaly_channel {
        Some(_channel) => {
            // Attempt a lightweight ServerInfo or similar RPC.
            // For now, channel existence indicates the connection was established.
            Ok("channel connected".to_string())
        }
        None => Err("channel not available".to_string()),
    }
}

/// Verify the job runner is alive by checking its heartbeat timestamp.
async fn check_job_runner(state: &AppState) -> Result<String, String> {
    // The job runner writes a heartbeat timestamp to a known MongoDB document.
    // If the heartbeat is older than 60 seconds, the runner is considered unhealthy.
    let _ = state;
    // Implementation: query `job_runner_heartbeat` document from MongoDB.
    // Placeholder -- will be wired when E06 job runner is available.
    Ok("heartbeat current".to_string())
}

use axum::{extract::Request, http::StatusCode, middleware::Next, response::Response};
use dashmap::DashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::response::{ApiError, ApiErrorBody};
use crate::request_context::RequestContext;

/// Configuration for the rate limiter.
#[derive(Debug, Clone)]
pub struct RateLimitConfig {
    /// Maximum requests per window per user.
    pub max_requests: u64,
    /// Window duration.
    pub window: Duration,
}

impl Default for RateLimitConfig {
    fn default() -> Self {
        Self {
            max_requests: 100,
            window: Duration::from_secs(60),
        }
    }
}

/// Per-user token bucket entry.
#[derive(Debug, Clone)]
struct BucketEntry {
    remaining: u64,
    window_start: Instant,
}

/// In-memory rate limit store. One entry per authenticated user.
///
/// Thread-safe via `DashMap`. Entries are lazily evicted when accessed
/// past their window.
#[derive(Debug, Clone)]
pub struct RateLimitStore {
    buckets: Arc<DashMap<String, BucketEntry>>,
    config: RateLimitConfig,
}

impl RateLimitStore {
    pub fn new(config: RateLimitConfig) -> Self {
        Self {
            buckets: Arc::new(DashMap::new()),
            config,
        }
    }

    /// Attempt to consume one token for the given user.
    /// Returns `Ok(remaining)` if allowed, `Err(())` if rate limited.
    pub fn check(&self, user_id: &str) -> Result<u64, ()> {
        let now = Instant::now();
        let mut entry = self.buckets.entry(user_id.to_string()).or_insert_with(|| {
            BucketEntry {
                remaining: self.config.max_requests,
                window_start: now,
            }
        });

        // Reset window if expired.
        if now.duration_since(entry.window_start) >= self.config.window {
            entry.remaining = self.config.max_requests;
            entry.window_start = now;
        }

        if entry.remaining == 0 {
            return Err(());
        }

        entry.remaining -= 1;
        Ok(entry.remaining)
    }
}

/// Rate limiting middleware.
///
/// Extracts the authenticated user ID from request extensions (set by the
/// auth middleware). Unauthenticated requests are not rate-limited here
/// (they are rejected by the auth middleware first).
///
/// Returns 429 Too Many Requests when the limit is exceeded.
pub async fn rate_limit_middleware(
    req: Request,
    next: Next,
    store: RateLimitStore,
) -> Response {
    // Extract user ID from auth context if present.
    let user_id = req
        .extensions()
        .get::<crate::auth::AuthUser>()
        .map(|u| u.user_id.to_string());

    if let Some(uid) = user_id {
        match store.check(&uid) {
            Ok(remaining) => {
                let mut response = next.run(req).await;
                // Attach rate limit headers for client awareness.
                if let Ok(val) = remaining.to_string().parse() {
                    response.headers_mut().insert("X-RateLimit-Remaining", val);
                }
                response
            }
            Err(()) => {
                metrics::counter!(
                    crate::metrics::RATE_LIMITED_TOTAL,
                    "user_id" => uid
                );

                let body = ApiError {
                    error: ApiErrorBody {
                        code: "rate_limited",
                        message: "Too many requests. Please wait and try again.".to_string(),
                        request_id: RequestContext::current_request_id(),
                    },
                };

                (StatusCode::TOO_MANY_REQUESTS, axum::Json(body)).into_response()
            }
        }
    } else {
        // No authenticated user -- skip rate limiting.
        next.run(req).await
    }
}

{
  "status": "ok",
  "version": "0.1.0",
  "components": [
    { "name": "mongodb", "status": "healthy", "detail": "ping: 2ms" },
    { "name": "gitaly", "status": "healthy", "detail": "channel connected" },
    { "name": "job_runner", "status": "healthy", "detail": "heartbeat current" }
  ]
}

{
  "status": "degraded",
  "version": "0.1.0",
  "components": [
    { "name": "mongodb", "status": "healthy", "detail": "ping: 3ms" },
    { "name": "gitaly", "status": "unhealthy", "detail": "channel not available" },
    { "name": "job_runner", "status": "healthy", "detail": "heartbeat current" }
  ]
}

use axum::response::IntoResponse;
use metrics_exporter_prometheus::PrometheusHandle;

/// GET /api/metrics
///
/// Returns metrics in Prometheus text exposition format. Not authenticated
/// -- restrict access via network policy in production.
pub async fn metrics_endpoint(
    State(handle): State<PrometheusHandle>,
) -> impl IntoResponse {
    handle.render()
}

{
  "error": {
    "code": "rate_limited",
    "message": "Too many requests. Please wait and try again.",
    "request_id": "018f2f35-2e63-7b3b-b5e1-9f0d3a2c4b10"
  }
}

// k6 script: tests/load/concurrent_edits.js
import http from 'k6/http';
import { check, sleep } from 'k6';

export const options = {
    vus: 50,
    duration: '2m',
    thresholds: {
        http_req_duration: ['p(99)<2000'],  // p99 < 2s
        http_req_failed: ['rate<0.01'],      // < 1% failure rate
    },
};

export default function () {
    const userId = __VU;
    const workspaceId = `ws-${userId}`;
    const fileName = `config/service-${userId}/settings.json`;

    const res = http.put(
        `${__ENV.BASE_URL}/api/apps/${__ENV.APP_ID}/workspaces/${workspaceId}/files`,
        JSON.stringify({ path: fileName, content: `{"vu": ${userId}, "iter": ${__ITER}}` }),
        { headers: { 'Authorization': `Bearer ${__ENV.TOKEN}`, 'Content-Type': 'application/json' } }
    );

    check(res, {
        'status is 200': (r) => r.status === 200,
        'response time < 2s': (r) => r.timings.duration < 2000,
    });

    sleep(1);
}

#[tokio::test]
async fn gitaly_down_returns_502_for_git_operations() {
    // Start mock gitaly that immediately drops connections.
    let mock_gitaly = MockGitalyServer::start_refusing_connections().await;
    let app = test_app_with_gitaly(mock_gitaly.address()).await;

    // Non-git endpoint should still work.
    let health_res = app
        .oneshot(Request::builder().uri("/api/health").body(Body::empty()).unwrap())
        .await
        .unwrap();
    // Health endpoint reports gitaly as unhealthy but responds 503 overall.
    assert_eq!(health_res.status(), StatusCode::SERVICE_UNAVAILABLE);
    let body: serde_json::Value = parse_body(health_res).await;
    assert_eq!(body["status"], "degraded");

    let gitaly_component = body["components"]
        .as_array()
        .unwrap()
        .iter()
        .find(|c| c["name"] == "gitaly")
        .unwrap();
    assert_eq!(gitaly_component["status"], "unhealthy");

    // Git operation should return 502.
    let file_res = app_clone
        .oneshot(
            Request::builder()
                .uri("/api/apps/test-app/workspaces/ws-1/files?path=config.json")
                .header("Authorization", "Bearer valid-token")
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(file_res.status(), StatusCode::BAD_GATEWAY);

    let err_body: serde_json::Value = parse_body(file_res).await;
    assert_eq!(err_body["error"]["code"], "git_error");
}

#[tokio::test]
async fn mongo_failover_recovers_automatically() {
    // Requires a 3-node replica set (testcontainers or local).
    let rs = MongoReplicaSet::start(3).await;
    let app = test_app_with_mongo(rs.connection_string()).await;

    // Verify initial connectivity.
    let res = app.clone().oneshot(health_request()).await.unwrap();
    assert_eq!(res.status(), StatusCode::OK);

    // Force primary step-down.
    rs.step_down_primary().await;

    // Requests may fail briefly during election.
    tokio::time::sleep(Duration::from_secs(2)).await;

    // After election completes, operations should resume.
    // Retry for up to 15 seconds.
    let mut recovered = false;
    for _ in 0..15 {
        let res = app.clone().oneshot(health_request()).await.unwrap();
        if res.status() == StatusCode::OK {
            recovered = true;
            break;
        }
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
    assert!(recovered, "Server did not recover after MongoDB failover within 15s");
}

#[tokio::test]
async fn crashed_job_is_retried_after_lock_expiry() {
    let db = test_mongo_db().await;
    let job_repo = JobRepo::new(db.clone());

    // Insert a job in "running" state with a stale lock (expired 1 minute ago).
    let stale_job = Job {
        id: ObjectId::new().to_hex(),
        job_type: "msuite_submit".to_string(),
        state: "running".to_string(),
        locked_until: Some(Utc::now() - Duration::from_secs(60)),
        attempts: 1,
        max_attempts: 3,
        ..test_job()
    };
    job_repo.insert(&stale_job).await.unwrap();

    // Start a new job runner instance.
    let runner = JobRunner::new(db.clone(), mock_workers());
    let picked = runner.poll_next_job().await.unwrap();

    // The stale job should be picked up for retry.
    assert!(picked.is_some());
    assert_eq!(picked.unwrap().id, stale_job.id);
    assert_eq!(picked.unwrap().attempts, 2);
}

#[tokio::test]
async fn nosql_injection_in_query_param_rejected() {
    let app = test_app().await;

    // Attempt MongoDB operator injection via query parameter.
    let res = app
        .oneshot(
            Request::builder()
                .uri("/api/apps?name[$gt]=")
                .header("Authorization", "Bearer valid-token")
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();

    // Should not return all apps -- either 400 (bad param) or empty results.
    // Must NOT return documents where name > "".
    assert!(
        res.status() == StatusCode::BAD_REQUEST || {
            let body: serde_json::Value = parse_body(res).await;
            body["data"].as_array().map_or(true, |arr| arr.is_empty())
        }
    );
}

#[tokio::test]
async fn nosql_injection_in_json_body_rejected() {
    let app = test_app().await;

    // Attempt operator injection in JSON body.
    // Serde's typed deserialization rejects objects where a String is expected.
    let res = app
        .oneshot(
            Request::builder()
                .method("POST")
                .uri("/api/apps")
                .header("Authorization", "Bearer admin-token")
                .header("Content-Type", "application/json")
                .body(Body::from(r#"{"name": {"$gt": ""}, "repo_path": "test.git"}"#))
                .unwrap(),
        )
        .await
        .unwrap();

    // Serde deserialization fails: name expects a string, not an object.
    assert_eq!(res.status(), StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn path_traversal_blocked() {
    let app = test_app_with_workspace().await;

    let traversal_paths = vec![
        "../../etc/passwd",
        "..%2F..%2Fetc%2Fpasswd",
        "config/../../../etc/shadow",
        "/etc/passwd",
        "config/../../.git/config",
    ];

    for path in traversal_paths {
        let res = app
            .clone()
            .oneshot(
                Request::builder()
                    .uri(&format!(
                        "/api/apps/test-app/workspaces/ws-1/files?path={}",
                        path
                    ))
                    .header("Authorization", "Bearer valid-token")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();

        assert!(
            res.status() == StatusCode::BAD_REQUEST || res.status() == StatusCode::FORBIDDEN,
            "Path traversal not blocked for: {path}"
        );
    }
}

#[tokio::test]
async fn expired_jwt_rejected() {
    let app = test_app().await;

    // Generate a JWT that expired 1 hour ago.
    let expired_token = issue_test_jwt(
        "user@example.com",
        Utc::now() - chrono::Duration::hours(1),
    );

    let res = app
        .oneshot(
            Request::builder()
                .uri("/api/apps")
                .header("Authorization", format!("Bearer {expired_token}"))
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();

    assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
    let body: serde_json::Value = parse_body(res).await;
    assert_eq!(body["error"]["code"], "unauthorized");
}

#[tokio::test]
async fn user_cannot_escalate_own_role() {
    let app = test_app_with_membership("user@test.com", Role::User).await;

    // Attempt to update own role to app_admin via the membership API.
    let res = app
        .oneshot(
            Request::builder()
                .method("PATCH")
                .uri("/api/apps/test-app/members/self")
                .header("Authorization", "Bearer user-token")
                .header("Content-Type", "application/json")
                .body(Body::from(r#"{"role": "app_admin"}"#))
                .unwrap(),
        )
        .await
        .unwrap();

    assert_eq!(res.status(), StatusCode::FORBIDDEN);
}

#[tokio::test]
async fn reviewer_cannot_manage_settings() {
    let app = test_app_with_membership("reviewer@test.com", Role::Reviewer).await;

    let res = app
        .oneshot(
            Request::builder()
                .method("PATCH")
                .uri("/api/apps/test-app/settings")
                .header("Authorization", "Bearer reviewer-token")
                .header("Content-Type", "application/json")
                .body(Body::from(r#"{"baseline_mode": "integration_head"}"#))
                .unwrap(),
        )
        .await
        .unwrap();

    assert_eq!(res.status(), StatusCode::FORBIDDEN);
}