http.rs

use std::{
    str::FromStr,
    sync::Arc,
    time::{Instant, SystemTime, UNIX_EPOCH},
};

#[cfg(feature = "tls")]
use rustls::{
    pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer},
    ServerConfig,
};

use actix_multipart::form::MultipartForm;
use actix_web::{
    web::{self, Data},
    App, HttpRequest, HttpResponse, HttpServer, Responder,
};
use actix_web_prometheus::PrometheusMetricsBuilder;
use agg_mode_sdk::{blockchain::AggregationModeProvingSystem, types::Network};
use alloy::signers::Signature;
use sp1_sdk::{SP1ProofWithPublicValues, SP1VerifyingKey};
use sqlx::types::BigDecimal;

use super::{
    helpers::format_merkle_path,
    types::{AppResponse, GetReceiptsQueryParams},
};

use crate::{
    config::Config,
    db::Db,
    helpers::get_time_left_day_formatted,
    metrics::GatewayMetrics,
    types::{GetReceiptsResponse, SubmitProofRequestRisc0, SubmitProofRequestSP1},
    verifiers::{verify_sp1_proof, VerificationError},
};

#[derive(Clone, Debug)]
pub struct GatewayServer {
    db: Db,
    config: Config,
    network: Network,
    metrics: Arc<GatewayMetrics>,
}

impl GatewayServer {
    pub fn new(db: Db, config: Config) -> Self {
        let network = Network::from_str(&config.network).expect("A valid network in config file");

        tracing::info!(
            "Starting metrics server on port {}",
            config.gateway_metrics_port
        );
        let metrics = GatewayMetrics::start(config.gateway_metrics_port)
            .expect("Failed to start metrics server");

        Self {
            db,
            config,
            network,
            metrics,
        }
    }

    #[cfg(feature = "tls")]
    fn load_tls_config(
        cert_path: &str,
        key_path: &str,
    ) -> Result<ServerConfig, Box<dyn std::error::Error>> {
        // Install the default crypto provider
        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();

        // Load certificate chain
        let certs: Vec<CertificateDer> =
            CertificateDer::pem_file_iter(cert_path)?.collect::<Result<Vec<_>, _>>()?;

        // Load private key
        let private_key = PrivateKeyDer::from_pem_file(key_path)?;

        let config = ServerConfig::builder()
            .with_no_client_auth()
            .with_single_cert(certs, private_key)?;

        Ok(config)
    }

    pub async fn start(&self) {
        // Note: GatewayServer is thread safe so we can just clone it (no need to add mutexes)
        let http_port = self.config.port;
        let state = self.clone();

        // Note: This creates a new Prometheus server different from the one created in GatewayServer::new. The created
        // server exposes metrics related to the actix HTTP server, like response codes and response times
        let prometheus = PrometheusMetricsBuilder::new("api")
            .endpoint("/metrics")
            .build()
            .unwrap();

        let server = HttpServer::new(move || {
            App::new()
                .app_data(Data::new(state.clone()))
                .wrap(prometheus.clone())
                .route("/", web::get().to(Self::get_root))
                .route("/nonce/{address}", web::get().to(Self::get_nonce))
                .route("/receipts", web::get().to(Self::get_receipts))
                .route("/proof/sp1", web::post().to(Self::post_proof_sp1))
                .route("/proof/risc0", web::post().to(Self::post_proof_risc0))
                .route("/quotas/{address}", web::get().to(Self::get_quotas))
        });

        #[cfg(feature = "tls")]
        {
            let tls_port = self.config.tls_port;
            tracing::info!(
                "Starting HTTP server at http://{}:{}",
                self.config.ip,
                http_port
            );
            tracing::info!(
                "Starting HTTPS server at https://{}:{}",
                self.config.ip,
                tls_port
            );

            let tls_config =
                Self::load_tls_config(&self.config.tls_cert_path, &self.config.tls_key_path)
                    .expect("Failed to load TLS configuration");

            server
                .bind((self.config.ip.as_str(), http_port))
                .expect("To bind HTTP socket correctly")
                .bind_rustls_0_23((self.config.ip.as_str(), tls_port), tls_config)
                .expect("To bind HTTPS socket correctly with TLS")
                .run()
                .await
                .expect("Server to never end");
        }

        #[cfg(not(feature = "tls"))]
        {
            tracing::info!(
                "Starting HTTP server at http://{}:{}",
                self.config.ip,
                http_port
            );

            server
                .bind((self.config.ip.as_str(), http_port))
                .expect("To bind HTTP socket correctly")
                .run()
                .await
                .expect("Server to never end");
        }
    }

    // Returns an OK response (code 200), no matters what receives in the request
    async fn get_root(_req: HttpRequest) -> impl Responder {
        HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({})))
    }

    // Returns the nonce (number of submitted tasks) for a given address
    async fn get_nonce(req: HttpRequest) -> impl Responder {
        let Some(address_raw) = req.match_info().get("address") else {
            return HttpResponse::BadRequest()
                .json(AppResponse::new_unsucessfull("Missing address", 400));
        };

        // Check that the address is a valid ethereum address
        if alloy::primitives::Address::from_str(address_raw.trim()).is_err() {
            return HttpResponse::BadRequest()
                .json(AppResponse::new_unsucessfull("Invalid address", 400));
        }

        let address = address_raw.to_lowercase();

        let Some(state) = req.app_data::<Data<GatewayServer>>() else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };

        let state = state.get_ref();
        match state.db.count_tasks_by_address(&address).await {
            Ok(count) => HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!(
                {
                    "nonce": count
                }
            ))),
            Err(_) => HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500)),
        }
    }

    // Posts an SP1 proof to the gateway, recovering the address from the signature
    async fn post_proof_sp1(
        req: HttpRequest,
        MultipartForm(data): MultipartForm<SubmitProofRequestSP1>,
    ) -> impl Responder {
        let Some(state) = req.app_data::<Data<GatewayServer>>() else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };

        let state = state.get_ref();
        let Ok(signature) = Signature::from_str(&data.signature_hex.0) else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Invalid signature", 500));
        };

        let Ok(proof_content) = tokio::fs::read(data.proof.file.path()).await else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };

        let Ok(vk_content) = tokio::fs::read(data.program_vk.file.path()).await else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };

        // reconstruct message and recover address
        let msg = agg_mode_sdk::gateway::types::SubmitSP1ProofMessage::new(
            data.nonce.0,
            proof_content.clone(),
            vk_content.clone(),
        );
        let Ok(recovered_address) =
            signature.recover_address_from_prehash(&msg.eip712_hash(&state.network).into())
        else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };
        let recovered_address = recovered_address.to_string().to_lowercase();

        // Checking if this address has submited more proofs than the ones allowed per day
        let Ok(daily_tasks_by_address) = state
            .db
            .get_daily_tasks_by_address(&recovered_address)
            .await
        else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };

        if daily_tasks_by_address >= state.config.max_daily_proofs_per_user {
            let formatted_time_left = get_time_left_day_formatted();

            return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(
                format!(
                    "Request denied: Query limit exceeded. Quotas renew in {formatted_time_left}"
                )
                .as_str(),
                400,
            ));
        }

        let Ok(count) = state.db.count_tasks_by_address(&recovered_address).await else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };

        if data.nonce.0 != (count as u64) {
            return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(
                &format!("Invalid nonce, expected nonce = {count}"),
                400,
            ));
        }

        let now_epoch = match SystemTime::now().duration_since(UNIX_EPOCH) {
            Ok(duration) => duration.as_secs(),
            Err(_) => {
                return HttpResponse::InternalServerError()
                    .json(AppResponse::new_unsucessfull("Internal server error", 500));
            }
        };

        let has_payment = match state
            .db
            .has_active_payment_event(
                &recovered_address,
                // safe unwrap the number comes from a valid u64 primitive
                BigDecimal::from_str(&now_epoch.to_string()).unwrap(),
            )
            .await
        {
            Ok(result) => result,
            Err(_) => {
                return HttpResponse::InternalServerError()
                    .json(AppResponse::new_unsucessfull("Internal server error", 500));
            }
        };

        if !has_payment {
            return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(
                "You have to pay before submitting a proof",
                400,
            ));
        }
        let Ok(proof) = bincode::deserialize::<SP1ProofWithPublicValues>(&proof_content) else {
            return HttpResponse::BadRequest()
                .json(AppResponse::new_unsucessfull("Invalid SP1 proof", 400));
        };

        let Ok(vk) = bincode::deserialize::<SP1VerifyingKey>(&vk_content) else {
            return HttpResponse::BadRequest()
                .json(AppResponse::new_unsucessfull("Invalid vk", 400));
        };

        if let Err(e) = verify_sp1_proof(&proof, &vk) {
            let message = match e {
                VerificationError::InvalidProof => "Proof verification failed",
                VerificationError::UnsupportedProof => "Unsupported proof",
            };

            return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(message, 400));
        };

        let query_started_at = Instant::now();

        match state
            .db
            .insert_task(
                &recovered_address,
                AggregationModeProvingSystem::SP1.as_u16() as i32,
                &proof_content,
                &vk_content,
                None,
                data.nonce.0 as i64,
            )
            .await
        {
            Ok(task_id) => {
                let time_elapsed_db_call = query_started_at.elapsed();
                state
                    .metrics
                    .register_db_response_time_post("sp1-post", time_elapsed_db_call.as_secs_f64());

                HttpResponse::Ok().json(AppResponse::new_sucessfull(
                    serde_json::json!({ "task_id": task_id.to_string() }),
                ))
            }
            Err(_) => HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500)),
        }
    }

    /// TODO: complete for risc0 (see `post_proof_sp1`)
    // Posts a Risc0 proof to the gateway, recovering the address from the signature
    async fn post_proof_risc0(
        _req: HttpRequest,
        MultipartForm(_): MultipartForm<SubmitProofRequestRisc0>,
    ) -> impl Responder {
        HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({})))
    }

    // Returns the last 100 receipt merkle proofs for the address received in the URL.
    // In case of also receiving a nonce on the query param, it returns only the merkle proof for that nonce.
    async fn get_receipts(
        req: HttpRequest,
        params: web::Query<GetReceiptsQueryParams>,
    ) -> impl Responder {
        let Some(state) = req.app_data::<Data<GatewayServer>>() else {
            return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
                "Internal server error: Failed to get app data",
                500,
            ));
        };

        let state = state.get_ref();

        if alloy::primitives::Address::from_str(params.address.clone().trim()).is_err() {
            return HttpResponse::BadRequest()
                .json(AppResponse::new_unsucessfull("Invalid address", 400));
        }

        let limit = match params.limit {
            Some(received_limit) => received_limit.min(100),
            None => 100,
        };

        let address = params.address.to_lowercase();

        let query = if let Some(nonce) = params.nonce {
            state
                .db
                .get_tasks_by_address_and_nonce(&address, nonce)
                .await
        } else {
            state
                .db
                .get_tasks_by_address_with_limit(&address, limit)
                .await
        };

        let Ok(receipts) = query else {
            return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
                "Internal server error: Failed to get tasks by address and nonce",
                500,
            ));
        };

        let responses: Result<Vec<GetReceiptsResponse>, String> = receipts
            .into_iter()
            .map(|receipt| {
                let Some(merkle_path) = receipt.merkle_path else {
                    return Ok(GetReceiptsResponse {
                        status: receipt.status,
                        merkle_path: Vec::new(),
                        nonce: receipt.nonce,
                        address: receipt.address,
                    });
                };

                let Ok(formatted) = format_merkle_path(&merkle_path) else {
                    return Err("Error formatting merkle path".into());
                };

                Ok(GetReceiptsResponse {
                    status: receipt.status,
                    merkle_path: formatted,
                    nonce: receipt.nonce,
                    address: receipt.address,
                })
            })
            .collect();

        match responses {
            Ok(resp) => HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({
                "receipts": resp
            }))),
            Err(_) => HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500)),
        }
    }

    async fn get_quotas(req: HttpRequest) -> impl Responder {
        let Some(state) = req.app_data::<Data<GatewayServer>>() else {
            return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
                "Internal server error: Failed to get app data",
                500,
            ));
        };

        let state = state.get_ref();

        let Some(address_raw) = req.match_info().get("address") else {
            return HttpResponse::BadRequest()
                .json(AppResponse::new_unsucessfull("Missing address", 400));
        };

        // Check that the address is a valid ethereum address
        if alloy::primitives::Address::from_str(address_raw.trim()).is_err() {
            return HttpResponse::BadRequest()
                .json(AppResponse::new_unsucessfull("Invalid address", 400));
        }

        let address = address_raw.trim().to_lowercase();

        let Ok(daily_tasks_by_address) = state.db.get_daily_tasks_by_address(&address).await else {
            return HttpResponse::InternalServerError()
                .json(AppResponse::new_unsucessfull("Internal server error", 500));
        };

        let formatted_time_left = get_time_left_day_formatted();

        let now_epoch = match SystemTime::now().duration_since(UNIX_EPOCH) {
            Ok(duration) => duration.as_secs(),
            Err(_) => {
                return HttpResponse::InternalServerError()
                    .json(AppResponse::new_unsucessfull("Internal server error", 500));
            }
        };

        let has_payment = match state
            .db
            .has_active_payment_event(
                &address,
                // safe unwrap the number comes from a valid u64 primitive
                BigDecimal::from_str(&now_epoch.to_string()).unwrap(),
            )
            .await
        {
            Ok(result) => result,
            Err(_) => {
                return HttpResponse::InternalServerError()
                    .json(AppResponse::new_unsucessfull("Internal server error", 500));
            }
        };

        if has_payment {
            HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({
                "proofs_submitted": daily_tasks_by_address,
                "quota_limit": state.config.max_daily_proofs_per_user,
                "quota_remaining": (state.config.max_daily_proofs_per_user - daily_tasks_by_address),
                "quota_resets_in": formatted_time_left.as_str()
            })))
        } else {
            HttpResponse::Ok().json(AppResponse::new_unsucessfull(
                "The address doesn't have an active subscription",
                404,
            ))
        }
    }
}