AkurAI-DNS / src/routes.rs

Latest change 1f985ee57c4d7762e35f4e0bd2fbbfe716bf42ee - feat(domains): audited domain lifecycle - certificate-ready state, transitions, events, customer-safe status (AKURAI-DNS-4) by agent

use std::sync::Arc;

use axum::{
    Json,
    extract::{Path, State},
    http::StatusCode,
    response::IntoResponse,
};
use rusqlite::params;
use serde_json::json;
use tokio::sync::Mutex;
use tracing::{info, warn};

use crate::models::{
    CreateDomainRequest, DomainOwnership, DomainResponse, DomainState, DomainStatusResponse,
    FailDomainRequest, VerifyDomainRequest,
};
use crate::schema;

pub type Db = Arc<Mutex<rusqlite::Connection>>;

/// Shared application state injected into every handler.
#[derive(Clone)]
pub struct AppState {
    pub db: Db,
}

// ── Tenant extraction ──────────────────────────────────────────────────

#[derive(Debug)]
struct Tenant {
    org_id: String,
    workspace_id: String,
}

/// Extract tenant identity from request headers.
/// Returns 401 when either header is missing or empty.
fn require_tenant(
    headers: &axum::http::HeaderMap,
) -> Result<Tenant, (StatusCode, Json<serde_json::Value>)> {
    let missing = |name: &str| {
        (
            StatusCode::UNAUTHORIZED,
            Json(json!({"error": format!("missing required header: {name}")})),
        )
    };

    let org_id = headers
        .get("x-org-id")
        .and_then(|v| v.to_str().ok())
        .filter(|v| !v.is_empty())
        .ok_or_else(|| missing("x-org-id"))?
        .to_string();

    let workspace_id = headers
        .get("x-workspace-id")
        .and_then(|v| v.to_str().ok())
        .filter(|v| !v.is_empty())
        .ok_or_else(|| missing("x-workspace-id"))?
        .to_string();

    Ok(Tenant {
        org_id,
        workspace_id,
    })
}

// ── Domain format validation ───────────────────────────────────────────

fn is_valid_domain(domain: &str) -> bool {
    if domain.is_empty() || domain.len() > 253 {
        return false;
    }
    // Each label: 1–63 chars, alphanumeric + hyphens, no leading/trailing hyphen.
    for label in domain.split('.') {
        if label.is_empty() || label.len() > 63 {
            return false;
        }
        if label.starts_with('-') || label.ends_with('-') {
            return false;
        }
        if !label.chars().all(|c| c.is_ascii_alphanumeric() || c == '-') {
            return false;
        }
    }
    true
}

// ── Shared row loading ──────────────────────────────────────────────────

fn row_to_domain(row: &rusqlite::Row) -> rusqlite::Result<DomainOwnership> {
    Ok(DomainOwnership {
        id: row.get(0)?,
        org_id: row.get(1)?,
        workspace_id: row.get(2)?,
        domain: row.get(3)?,
        state: DomainState::from_str(&row.get::<_, String>(4)?).unwrap_or(DomainState::Requested),
        verification_token: row.get(5)?,
        verified_at: row.get(6)?,
        provisioned_at: row.get(7)?,
        cert_ready_at: row.get(8)?,
        expires_at: row.get(9)?,
        last_error: row.get(10)?,
        created_at: row.get(11)?,
        updated_at: row.get(12)?,
    })
}

const DOMAIN_COLUMNS: &str = "id, org_id, workspace_id, domain, state, verification_token,
     verified_at, provisioned_at, cert_ready_at, expires_at, last_error, created_at, updated_at";

/// Load a domain scoped to the caller's tenant. `None` covers both
/// "does not exist" and "belongs to a different tenant" — callers must
/// return 404 either way so cross-tenant existence is never leaked.
fn load_scoped(
    conn: &rusqlite::Connection,
    id: i64,
    org_id: &str,
    workspace_id: &str,
) -> Option<DomainOwnership> {
    conn.query_row(
        &format!(
            "SELECT {DOMAIN_COLUMNS} FROM domain_ownerships
             WHERE id = ?1 AND org_id = ?2 AND workspace_id = ?3"
        ),
        params![id, org_id, workspace_id],
        row_to_domain,
    )
    .ok()
}

fn emit_lifecycle_event(
    conn: &rusqlite::Connection,
    domain_id: i64,
    org_id: &str,
    workspace_id: &str,
    from_state: Option<&str>,
    to_state: &str,
    detail: Option<&str>,
) {
    if let Err(e) = conn.execute(
        "INSERT INTO domain_lifecycle_events
            (domain_id, org_id, workspace_id, from_state, to_state, detail)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
        params![
            domain_id,
            org_id,
            workspace_id,
            from_state,
            to_state,
            detail
        ],
    ) {
        // Auditing must never block the mutation it describes; the state
        // change itself already committed by the time this runs.
        warn!(error = %e, domain_id, to_state, "failed to record lifecycle event");
    }
}

fn not_found() -> (StatusCode, Json<serde_json::Value>) {
    (
        StatusCode::NOT_FOUND,
        Json(json!({"error": "domain ownership not found"})),
    )
}

fn conflict(msg: &str) -> (StatusCode, Json<serde_json::Value>) {
    (StatusCode::CONFLICT, Json(json!({"error": msg})))
}

// ── Handlers ───────────────────────────────────────────────────────────

/// POST /api/domains — register a new domain ownership.
pub async fn create_domain(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Json(body): Json<CreateDomainRequest>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let domain = body.domain.trim().to_lowercase();

    if !is_valid_domain(&domain) {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "invalid domain format"})),
        );
    }

    let token = uuid::Uuid::new_v4().to_string();
    let db = state.db.lock().await;

    let result = db.execute(
        "INSERT INTO domain_ownerships (org_id, workspace_id, domain, state, verification_token)
         VALUES (?1, ?2, ?3, 'requested', ?4)",
        params![tenant.org_id, tenant.workspace_id, domain, token],
    );

    match result {
        Ok(_) => {
            let id = db.last_insert_rowid();
            info!(org_id = %tenant.org_id, workspace_id = %tenant.workspace_id, domain = %domain, id = id, "domain ownership created");
            emit_lifecycle_event(
                &db,
                id,
                &tenant.org_id,
                &tenant.workspace_id,
                None,
                "requested",
                None,
            );
            let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
            let response = row.map(DomainResponse::from);
            (StatusCode::CREATED, Json(json!(response)))
        }
        Err(e) => {
            if e.to_string().contains("UNIQUE constraint") {
                return (
                    StatusCode::CONFLICT,
                    Json(json!({"error": "domain already registered"})),
                );
            }
            warn!(error = %e, "failed to insert domain ownership");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "failed to create domain ownership"})),
            )
        }
    }
}

/// GET /api/domains — list domains belonging to the caller's workspace.
pub async fn list_domains(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let db = state.db.lock().await;

    let mut stmt = match db.prepare(&format!(
        "SELECT {DOMAIN_COLUMNS} FROM domain_ownerships
         WHERE org_id = ?1 AND workspace_id = ?2
         ORDER BY created_at DESC"
    )) {
        Ok(s) => s,
        Err(e) => {
            warn!(error = %e, "failed to prepare list query");
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "internal server error"})),
            );
        }
    };
    let rows: Vec<DomainOwnership> =
        match stmt.query_map(params![tenant.org_id, tenant.workspace_id], row_to_domain) {
            Ok(mapped) => mapped.filter_map(|r| r.ok()).collect(),
            Err(_) => Vec::new(),
        };

    let response: Vec<DomainResponse> = rows.into_iter().map(DomainResponse::from).collect();
    (StatusCode::OK, Json(json!(response)))
}

/// GET /api/domains/{id}/status — customer-safe status view (no
/// verification token, no zone internals) for Platform display.
pub async fn get_domain_status(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(id): Path<i64>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let db = state.db.lock().await;
    match load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id) {
        Some(d) => (StatusCode::OK, Json(json!(DomainStatusResponse::from(d)))),
        None => not_found(),
    }
}

/// POST /api/domains/{id}/verify — complete ownership verification and
/// provision the domain. Idempotent: replaying with the same token once
/// already provisioned (or further along) returns the current state
/// without re-mutating.
pub async fn verify_domain(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(id): Path<i64>,
    Json(body): Json<VerifyDomainRequest>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let db = state.db.lock().await;
    let Some(d) = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id) else {
        return not_found();
    };

    let Some(expected) = &d.verification_token else {
        return conflict("domain has no pending verification token");
    };
    if expected != &body.token {
        return (
            StatusCode::FORBIDDEN,
            Json(json!({"error": "verification token mismatch"})),
        );
    }

    // Already provisioned (or further along) with a matching token: the
    // caller is replaying a prior success, not requesting a new mutation.
    if !matches!(d.state, DomainState::Requested | DomainState::Verifying) {
        let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
        return (StatusCode::OK, Json(json!(row.map(DomainResponse::from))));
    }

    let from = d.state.as_str();
    let result = db.execute(
        "UPDATE domain_ownerships
         SET state = 'provisioned', verified_at = datetime('now'),
             provisioned_at = datetime('now'), last_error = NULL,
             updated_at = datetime('now')
         WHERE id = ?1",
        params![id],
    );
    match result {
        Ok(_) => {
            emit_lifecycle_event(
                &db,
                id,
                &tenant.org_id,
                &tenant.workspace_id,
                Some(from),
                "provisioned",
                None,
            );
            let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
            (StatusCode::OK, Json(json!(row.map(DomainResponse::from))))
        }
        Err(e) => {
            warn!(error = %e, id, "failed to verify domain");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "failed to verify domain"})),
            )
        }
    }
}

/// POST /api/domains/{id}/cert-ready — record that a TLS certificate has
/// been issued for a provisioned (or renewing) domain.
pub async fn mark_cert_ready(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(id): Path<i64>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let db = state.db.lock().await;
    let Some(d) = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id) else {
        return not_found();
    };

    if d.state == DomainState::CertificateReady {
        let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
        return (StatusCode::OK, Json(json!(row.map(DomainResponse::from))));
    }
    if !matches!(d.state, DomainState::Provisioned | DomainState::Renewing) {
        return conflict("domain must be provisioned or renewing before its certificate is ready");
    }

    let from = d.state.as_str();
    let result = db.execute(
        "UPDATE domain_ownerships
         SET state = 'certificate_ready', cert_ready_at = datetime('now'),
             last_error = NULL, updated_at = datetime('now')
         WHERE id = ?1",
        params![id],
    );
    match result {
        Ok(_) => {
            emit_lifecycle_event(
                &db,
                id,
                &tenant.org_id,
                &tenant.workspace_id,
                Some(from),
                "certificate_ready",
                None,
            );
            let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
            (StatusCode::OK, Json(json!(row.map(DomainResponse::from))))
        }
        Err(e) => {
            warn!(error = %e, id, "failed to mark certificate ready");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "failed to mark certificate ready"})),
            )
        }
    }
}

/// POST /api/domains/{id}/renew — begin a renewal cycle for a
/// certificate-ready domain. Completed via the same `/cert-ready` endpoint.
pub async fn renew_domain(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(id): Path<i64>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let db = state.db.lock().await;
    let Some(d) = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id) else {
        return not_found();
    };

    if d.state == DomainState::Renewing {
        let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
        return (StatusCode::OK, Json(json!(row.map(DomainResponse::from))));
    }
    if d.state != DomainState::CertificateReady {
        return conflict("domain must be certificate-ready before it can be renewed");
    }

    let from = d.state.as_str();
    let result = db.execute(
        "UPDATE domain_ownerships SET state = 'renewing', updated_at = datetime('now') WHERE id = ?1",
        params![id],
    );
    match result {
        Ok(_) => {
            emit_lifecycle_event(
                &db,
                id,
                &tenant.org_id,
                &tenant.workspace_id,
                Some(from),
                "renewing",
                None,
            );
            let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
            (StatusCode::OK, Json(json!(row.map(DomainResponse::from))))
        }
        Err(e) => {
            warn!(error = %e, id, "failed to start renewal");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "failed to start renewal"})),
            )
        }
    }
}

/// POST /api/domains/{id}/fail — record a recoverable lifecycle failure.
pub async fn fail_domain(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(id): Path<i64>,
    Json(body): Json<FailDomainRequest>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let db = state.db.lock().await;
    let Some(d) = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id) else {
        return not_found();
    };
    if d.state == DomainState::Retired {
        return conflict("a retired domain cannot transition to failed");
    }

    let from = d.state.as_str();
    let reason = body.reason.trim();
    let reason = if reason.is_empty() {
        "unspecified failure"
    } else {
        reason
    };
    let result = db.execute(
        "UPDATE domain_ownerships
         SET state = 'failed', last_error = ?2, updated_at = datetime('now')
         WHERE id = ?1",
        params![id, reason],
    );
    match result {
        Ok(_) => {
            emit_lifecycle_event(
                &db,
                id,
                &tenant.org_id,
                &tenant.workspace_id,
                Some(from),
                "failed",
                Some(reason),
            );
            let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
            (StatusCode::OK, Json(json!(row.map(DomainResponse::from))))
        }
        Err(e) => {
            warn!(error = %e, id, "failed to record domain failure");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "failed to record domain failure"})),
            )
        }
    }
}

/// POST /api/domains/{id}/retry — retry a failed domain, returning it to
/// `requested`. Requires an `Idempotency-Key` header; replaying the same
/// key returns the (already-applied) current state without re-mutating.
pub async fn retry_domain(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(id): Path<i64>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let Some(idempotency_key) = headers
        .get("idempotency-key")
        .and_then(|v| v.to_str().ok())
        .filter(|v| !v.is_empty())
    else {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "missing required header: Idempotency-Key"})),
        );
    };

    let db = state.db.lock().await;
    let Some(d) = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id) else {
        return not_found();
    };

    let claimed = db.execute(
        "INSERT INTO domain_retry_idempotency (domain_id, idempotency_key) VALUES (?1, ?2)",
        params![id, idempotency_key],
    );
    if claimed.is_err() {
        // Key already used for this domain — this is a replay of a request
        // already applied (or currently being applied); return the current
        // state rather than mutating again.
        let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
        return (StatusCode::OK, Json(json!(row.map(DomainResponse::from))));
    }

    if d.state != DomainState::Failed {
        return conflict("only a failed domain can be retried");
    }

    let result = db.execute(
        "UPDATE domain_ownerships
         SET state = 'requested', last_error = NULL, updated_at = datetime('now')
         WHERE id = ?1",
        params![id],
    );
    match result {
        Ok(_) => {
            emit_lifecycle_event(
                &db,
                id,
                &tenant.org_id,
                &tenant.workspace_id,
                Some("failed"),
                "requested",
                Some("retry"),
            );
            let row = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id);
            (StatusCode::OK, Json(json!(row.map(DomainResponse::from))))
        }
        Err(e) => {
            warn!(error = %e, id, "failed to retry domain");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "failed to retry domain"})),
            )
        }
    }
}

/// DELETE /api/domains/{id} — retire a domain ownership (tenant-scoped).
/// Soft: transitions to `retired` rather than deleting the row, so the
/// full lifecycle (including retirement) stays auditable.
pub async fn delete_domain(
    State(state): State<AppState>,
    headers: axum::http::HeaderMap,
    Path(id): Path<i64>,
) -> impl IntoResponse {
    let tenant = match require_tenant(&headers) {
        Ok(t) => t,
        Err(resp) => return resp,
    };

    let db = state.db.lock().await;
    let Some(d) = load_scoped(&db, id, &tenant.org_id, &tenant.workspace_id) else {
        return not_found();
    };
    if d.state == DomainState::Retired {
        return (StatusCode::NO_CONTENT, Json(json!(null)));
    }

    let from = d.state.as_str();
    match db.execute(
        "UPDATE domain_ownerships SET state = 'retired', updated_at = datetime('now') WHERE id = ?1",
        params![id],
    ) {
        Ok(1) => {
            info!(org_id = %tenant.org_id, workspace_id = %tenant.workspace_id, id = id, "domain ownership retired");
            emit_lifecycle_event(
                &db,
                id,
                &tenant.org_id,
                &tenant.workspace_id,
                Some(from),
                "retired",
                None,
            );
            (StatusCode::NO_CONTENT, Json(json!(null)))
        }
        Ok(_) => (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": "unexpected retire result"})),
        ),
        Err(e) => {
            warn!(error = %e, "failed to retire domain ownership");
            (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": "failed to retire domain ownership"})),
            )
        }
    }
}

/// Initialize the database and run migrations.
pub fn init_db(path: &str) -> Result<Db, Box<dyn std::error::Error>> {
    let conn = rusqlite::Connection::open(path)?;
    conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA foreign_keys=ON;")?;
    schema::migrate(&conn)?;
    Ok(Arc::new(Mutex::new(conn)))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn valid_domains_pass() {
        assert!(is_valid_domain("example.com"));
        assert!(is_valid_domain("sub.example.com"));
        assert!(is_valid_domain("a.io"));
    }

    #[test]
    fn invalid_domains_rejected() {
        assert!(!is_valid_domain(""));
        assert!(!is_valid_domain("-example.com"));
        assert!(!is_valid_domain("example-.com"));
    }

    // ── Handler-level lifecycle tests (in-memory DB, no HTTP layer) ──────

    fn test_state() -> AppState {
        let conn = rusqlite::Connection::open_in_memory().unwrap();
        schema::migrate(&conn).unwrap();
        AppState {
            db: Arc::new(Mutex::new(conn)),
        }
    }

    fn headers_for(org: &str, ws: &str) -> axum::http::HeaderMap {
        let mut h = axum::http::HeaderMap::new();
        h.insert("x-org-id", org.parse().unwrap());
        h.insert("x-workspace-id", ws.parse().unwrap());
        h
    }

    async fn create(state: &AppState, org: &str, ws: &str, domain: &str) -> DomainOwnership {
        let db = state.db.lock().await;
        db.execute(
            "INSERT INTO domain_ownerships (org_id, workspace_id, domain, state, verification_token)
             VALUES (?1, ?2, ?3, 'requested', 'tok-abc')",
            params![org, ws, domain],
        )
        .unwrap();
        let id = db.last_insert_rowid();
        load_scoped(&db, id, org, ws).unwrap()
    }

    #[tokio::test]
    async fn full_lifecycle_verify_cert_ready_renew_cert_ready_again() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let headers = headers_for("org-1", "ws-1");

        let resp = verify_domain(
            State(state.clone()),
            headers.clone(),
            Path(d.id),
            Json(VerifyDomainRequest {
                token: "tok-abc".into(),
            }),
        )
        .await;
        assert_eq!(resp.into_response().status(), StatusCode::OK);

        let resp = mark_cert_ready(State(state.clone()), headers.clone(), Path(d.id)).await;
        assert_eq!(resp.into_response().status(), StatusCode::OK);

        let resp = renew_domain(State(state.clone()), headers.clone(), Path(d.id)).await;
        assert_eq!(resp.into_response().status(), StatusCode::OK);

        let resp = mark_cert_ready(State(state.clone()), headers.clone(), Path(d.id)).await;
        assert_eq!(resp.into_response().status(), StatusCode::OK);

        let db = state.db.lock().await;
        let final_row = load_scoped(&db, d.id, "org-1", "ws-1").unwrap();
        assert_eq!(final_row.state, DomainState::CertificateReady);
        assert!(final_row.cert_ready_at.is_some());
    }

    #[tokio::test]
    async fn verify_rejects_wrong_token() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let resp = verify_domain(
            State(state.clone()),
            headers_for("org-1", "ws-1"),
            Path(d.id),
            Json(VerifyDomainRequest {
                token: "wrong".into(),
            }),
        )
        .await;
        assert_eq!(resp.into_response().status(), StatusCode::FORBIDDEN);
    }

    #[tokio::test]
    async fn verify_is_idempotent_once_already_provisioned() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let headers = headers_for("org-1", "ws-1");
        let req = || {
            Json(VerifyDomainRequest {
                token: "tok-abc".into(),
            })
        };

        let first = verify_domain(State(state.clone()), headers.clone(), Path(d.id), req()).await;
        assert_eq!(first.into_response().status(), StatusCode::OK);

        // Replaying the exact same verify call must not error and must not
        // regress the state.
        let second = verify_domain(State(state.clone()), headers.clone(), Path(d.id), req()).await;
        assert_eq!(second.into_response().status(), StatusCode::OK);

        let db = state.db.lock().await;
        let row = load_scoped(&db, d.id, "org-1", "ws-1").unwrap();
        assert_eq!(row.state, DomainState::Provisioned);
    }

    #[tokio::test]
    async fn fail_then_retry_returns_to_requested_and_is_idempotent() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let headers = headers_for("org-1", "ws-1");

        let resp = fail_domain(
            State(state.clone()),
            headers.clone(),
            Path(d.id),
            Json(FailDomainRequest {
                reason: "DNS propagation timeout".into(),
            }),
        )
        .await;
        assert_eq!(resp.into_response().status(), StatusCode::OK);

        let mut retry_headers = headers.clone();
        retry_headers.insert("idempotency-key", "retry-key-1".parse().unwrap());

        let first = retry_domain(State(state.clone()), retry_headers.clone(), Path(d.id)).await;
        assert_eq!(first.into_response().status(), StatusCode::OK);

        {
            let db = state.db.lock().await;
            let row = load_scoped(&db, d.id, "org-1", "ws-1").unwrap();
            assert_eq!(row.state, DomainState::Requested);
            assert!(row.last_error.is_none());
        }

        // A second retry attempt with a NEW idempotency key on an
        // already-requested (not failed) domain must be rejected...
        let mut retry_headers_2 = headers.clone();
        retry_headers_2.insert("idempotency-key", "retry-key-2".parse().unwrap());
        let second = retry_domain(State(state.clone()), retry_headers_2, Path(d.id)).await;
        assert_eq!(second.into_response().status(), StatusCode::CONFLICT);

        // ...but replaying the ORIGINAL key is a no-op success, not an error.
        let replay = retry_domain(State(state.clone()), retry_headers, Path(d.id)).await;
        assert_eq!(replay.into_response().status(), StatusCode::OK);
    }

    #[tokio::test]
    async fn retry_requires_idempotency_key_header() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let resp = retry_domain(
            State(state.clone()),
            headers_for("org-1", "ws-1"),
            Path(d.id),
        )
        .await;
        assert_eq!(resp.into_response().status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn cross_tenant_cannot_see_or_mutate_another_workspaces_domain() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let other = headers_for("org-2", "ws-2");

        let status = get_domain_status(State(state.clone()), other.clone(), Path(d.id)).await;
        assert_eq!(status.into_response().status(), StatusCode::NOT_FOUND);

        let verify = verify_domain(
            State(state.clone()),
            other.clone(),
            Path(d.id),
            Json(VerifyDomainRequest {
                token: "tok-abc".into(),
            }),
        )
        .await;
        assert_eq!(verify.into_response().status(), StatusCode::NOT_FOUND);

        let mut retry_headers = other;
        retry_headers.insert("idempotency-key", "x".parse().unwrap());
        let delete = delete_domain(State(state.clone()), retry_headers, Path(d.id)).await;
        assert_eq!(delete.into_response().status(), StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn delete_retires_rather_than_erasing_the_row() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let headers = headers_for("org-1", "ws-1");

        let resp = delete_domain(State(state.clone()), headers.clone(), Path(d.id)).await;
        assert_eq!(resp.into_response().status(), StatusCode::NO_CONTENT);

        let db = state.db.lock().await;
        let row = load_scoped(&db, d.id, "org-1", "ws-1");
        assert!(row.is_some(), "retired domain must still exist for audit");
        assert_eq!(row.unwrap().state, DomainState::Retired);

        let events: i64 = db
            .query_row(
                "SELECT COUNT(*) FROM domain_lifecycle_events WHERE domain_id = ?1 AND to_state = 'retired'",
                params![d.id],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(events, 1);
    }

    #[tokio::test]
    async fn status_view_never_exposes_the_verification_token() {
        let state = test_state();
        let d = create(&state, "org-1", "ws-1", "example.com").await;
        let db = state.db.lock().await;
        let row = load_scoped(&db, d.id, "org-1", "ws-1").unwrap();
        let status = DomainStatusResponse::from(row);
        let json = serde_json::to_string(&status).unwrap();
        assert!(!json.contains("tok-abc"));
        assert!(!json.contains("verification_token"));
    }
}