hypercall_api/handlers/

health.rs

use crate::runtime_status::{StandbyReplayProgress, StartupProgressReader};
use crate::sonic_json::SonicJson;
use axum::{extract::State, http::StatusCode, response::IntoResponse};
use serde_json::json;

use super::AppState;

pub use hypercall_runtime_api::drain::drain_marker_path;

/// Health check endpoint
#[utoipa::path(
    get,
    path = "/health",
    responses(
        (status = 200, description = "Health status", body = HealthResponse),
        (status = 503, description = "Service is shutting down", body = HealthResponse)
    ),
    tag = "Health"
)]
pub async fn health(State(app_state): State<AppState>) -> impl IntoResponse {
    let shutting_down = app_state.shutdown.is_triggered();
    let response = crate::models::HealthResponse {
        status: if shutting_down {
            "shutting_down".to_string()
        } else {
            "ok".to_string()
        },
    };

    if shutting_down {
        (StatusCode::SERVICE_UNAVAILABLE, SonicJson(response))
    } else {
        (StatusCode::OK, SonicJson(response))
    }
}

/// Version information endpoint
#[utoipa::path(
    get,
    path = "/version",
    responses(
        (status = 200, description = "Version information", body = VersionResponse)
    ),
    tag = "Health"
)]
pub async fn version(
    State(app_state): State<AppState>,
) -> SonicJson<crate::models::VersionResponse> {
    let build_info = &app_state.build_info;
    SonicJson(crate::models::VersionResponse {
        version: build_info.version.clone(),
        commit: build_info.commit.clone(),
        git_ref: build_info.git_ref.clone(),
        build_time: build_info.build_time.clone(),
        signing_chain_id: Some(app_state.runtime_config.signing_chain_id),
    })
}

/// Readiness check endpoint
///
/// Returns 200 OK when all components are ready to serve requests,
/// or 503 Service Unavailable when the service is still starting up.
#[utoipa::path(
    get,
    path = "/ready",
    responses(
        (status = 200, description = "Service is ready", body = ReadyResponse),
        (status = 503, description = "Service is not ready", body = ReadyResponse)
    ),
    tag = "Health"
)]
pub async fn ready(State(app_state): State<AppState>) -> impl IntoResponse {
    let components = app_state.readiness.reports();
    let all_ready = app_state.readiness.all_ready();

    let response = crate::models::ReadyResponse {
        status: if all_ready {
            "ready".to_string()
        } else {
            "not_ready".to_string()
        },
        message: if all_ready {
            None
        } else {
            Some("Service is starting up. Please retry shortly.".to_string())
        },
        components,
    };

    if all_ready {
        (StatusCode::OK, SonicJson(response))
    } else {
        (StatusCode::SERVICE_UNAVAILABLE, SonicJson(response))
    }
}

/// Standby readiness — returns 200 when the replay loop has caught up with the primary.
pub async fn standby_ready(State(app_state): State<AppState>) -> axum::response::Response {
    let already_promoted = standby_already_promoted(app_state.standby_promote.as_ref());
    standby_ready_response_with_startup(
        app_state.standby_progress.as_deref(),
        already_promoted,
        app_state.startup_progress.as_deref(),
    )
}

pub fn standby_already_promoted(
    standby_promote: Option<
        &std::sync::Arc<tokio::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>>,
    >,
) -> bool {
    standby_promote.is_none_or(|p| {
        // If promote_tx has been taken, we're already promoted
        p.try_lock().is_ok_and(|guard| guard.is_none())
    })
}

pub fn standby_ready_response_with_startup(
    progress: Option<&dyn StandbyReplayProgress>,
    already_promoted: bool,
    startup_progress: Option<&dyn StartupProgressReader>,
) -> axum::response::Response {
    match progress {
        None => (
            StatusCode::NOT_FOUND,
            SonicJson(with_startup_progress(
                json!({"status": "not_standby_mode"}),
                startup_progress,
            )),
        )
            .into_response(),
        Some(progress) => {
            let response = with_startup_progress(
                standby_status_payload(progress, already_promoted),
                startup_progress,
            );

            if already_promoted {
                (StatusCode::CONFLICT, SonicJson(response)).into_response()
            } else if progress.is_caught_up() {
                (StatusCode::OK, SonicJson(response)).into_response()
            } else {
                (StatusCode::SERVICE_UNAVAILABLE, SonicJson(response)).into_response()
            }
        }
    }
}

fn with_startup_progress(
    mut payload: serde_json::Value,
    startup_progress: Option<&dyn StartupProgressReader>,
) -> serde_json::Value {
    let Some(startup_progress) = startup_progress else {
        return payload;
    };

    let snapshot = startup_progress.snapshot();
    payload["startup_phase"] = json!(snapshot.phase);
    payload["startup_progress_counter"] = json!(snapshot.counter);
    payload["last_startup_progress_unix_ms"] = json!(snapshot.last_progress_unix_ms);
    payload["last_startup_progress_age_ms"] = json!(snapshot.last_progress_age_ms);
    payload
}

pub fn standby_status_payload(
    progress: &dyn StandbyReplayProgress,
    already_promoted: bool,
) -> serde_json::Value {
    let caught_up = progress.is_caught_up();
    let last_replay_unix_ms = progress.last_replay_unix_ms();
    let last_replay_age_ms = last_replay_unix_ms.map(|ts| {
        let now_ms = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("system clock before UNIX_EPOCH")
            .as_millis() as u64;
        now_ms.saturating_sub(ts)
    });

    json!({
        "status": if already_promoted {
            "already_promoted"
        } else if caught_up {
            "standby_ready"
        } else {
            "catching_up"
        },
        "commands_replayed": progress.commands_replayed(),
        "caught_up": caught_up,
        "promotable": caught_up && !already_promoted,
        "replay_cursor_seq": progress.replay_cursor_seq(),
        "latest_stream_seq": progress.latest_stream_seq(),
        "stream_lag": progress.stream_lag(),
        "last_replayed_seq": progress.last_replayed_seq(),
        "last_replay_unix_ms": last_replay_unix_ms,
        "last_replay_age_ms": last_replay_age_ms,
    })
}

#[cfg(test)]
mod tests {
    use super::standby_status_payload;
    use crate::runtime_status::StandbyReplayProgress;

    #[derive(Default)]
    struct TestStandbyReplayProgress {
        commands_replayed: u64,
        caught_up: bool,
        replay_cursor_seq: Option<i64>,
        latest_stream_seq: Option<u64>,
        stream_lag: Option<u64>,
        last_replayed_seq: Option<i64>,
        last_replay_unix_ms: Option<u64>,
    }

    impl TestStandbyReplayProgress {
        fn record_replayed(&mut self, stream_seq: i64) {
            self.commands_replayed += 1;
            self.replay_cursor_seq = Some(stream_seq);
            self.latest_stream_seq = Some(stream_seq as u64);
            self.stream_lag = Some(0);
            self.last_replayed_seq = Some(stream_seq);
            self.last_replay_unix_ms = Some(
                std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .expect("system clock before UNIX_EPOCH")
                    .as_millis() as u64,
            );
        }

        fn set_caught_up(&mut self) {
            self.caught_up = true;
        }
    }

    impl StandbyReplayProgress for TestStandbyReplayProgress {
        fn commands_replayed(&self) -> u64 {
            self.commands_replayed
        }

        fn is_caught_up(&self) -> bool {
            self.caught_up
        }

        fn replay_cursor_seq(&self) -> Option<i64> {
            self.replay_cursor_seq
        }

        fn latest_stream_seq(&self) -> Option<u64> {
            self.latest_stream_seq
        }

        fn stream_lag(&self) -> Option<u64> {
            self.stream_lag
        }

        fn last_replayed_seq(&self) -> Option<i64> {
            self.last_replayed_seq
        }

        fn last_replay_unix_ms(&self) -> Option<u64> {
            self.last_replay_unix_ms
        }
    }

    #[test]
    fn standby_status_payload_reports_catching_up_progress() {
        let progress = TestStandbyReplayProgress::default();

        let payload = standby_status_payload(&progress, false);

        assert_eq!(payload["status"], "catching_up");
        assert_eq!(payload["commands_replayed"], 0);
        assert_eq!(payload["caught_up"], false);
        assert_eq!(payload["promotable"], false);
        assert!(payload["replay_cursor_seq"].is_null());
        assert!(payload["latest_stream_seq"].is_null());
        assert!(payload["stream_lag"].is_null());
        assert!(payload["last_replayed_seq"].is_null());
        assert!(payload["last_replay_unix_ms"].is_null());
        assert!(payload["last_replay_age_ms"].is_null());
    }

    #[test]
    fn standby_status_payload_reports_recent_replay_metadata() {
        let mut progress = TestStandbyReplayProgress::default();
        progress.record_replayed(42);
        progress.set_caught_up();

        let payload = standby_status_payload(&progress, false);

        assert_eq!(payload["status"], "standby_ready");
        assert_eq!(payload["commands_replayed"], 1);
        assert_eq!(payload["caught_up"], true);
        assert_eq!(payload["promotable"], true);
        assert_eq!(payload["replay_cursor_seq"], 42);
        assert_eq!(payload["latest_stream_seq"], 42);
        assert_eq!(payload["stream_lag"], 0);
        assert_eq!(payload["last_replayed_seq"], 42);
        assert!(payload["last_replay_unix_ms"].as_u64().is_some());
        assert!(payload["last_replay_age_ms"].as_u64().is_some());
    }

    #[test]
    fn standby_status_payload_reports_already_promoted() {
        let mut progress = TestStandbyReplayProgress::default();
        progress.record_replayed(99);
        progress.set_caught_up();

        let payload = standby_status_payload(&progress, true);

        assert_eq!(payload["status"], "already_promoted");
        assert_eq!(payload["promotable"], false);
    }
}

/// Public exchange configuration for frontend integration.
#[utoipa::path(
    get,
    path = "/exchange-info",
    responses(
        (status = 200, description = "Exchange configuration", body = ExchangeInfoResponse)
    ),
    tag = "Health"
)]
pub async fn exchange_info(
    State(app_state): State<AppState>,
) -> SonicJson<crate::models::ExchangeInfoResponse> {
    SonicJson(crate::models::ExchangeInfoResponse {
        exchange_address: app_state.runtime_config.exchange_contract_address.clone(),
        chain_id: app_state.runtime_config.signing_chain_id,
        signing_domain: crate::models::SigningDomainInfo {
            name: "Hypercall".to_string(),
            version: "1".to_string(),
        },
    })
}