nest_data_source_api/

api.rs

use crate::api::ApiError::Internal;
use crate::batch_messages::{
    ActivityBatchRequestMessageType, ActivityBatchResponseMessageType, HasPEPBatchInfo,
    PEPBatchMessageType, PEPPseudoDataPoint, ParticipantBatchRequestMessageType,
    ParticipantBatchResponseMessageType, PseudoDataPoint,
};
use crate::data_source::DataSource;
use crate::messages::{
    ActivityRequestMessageType, ActivityResponseMessageType, HasPEPParticipantInfo, PEPMessageType,
    ParticipantRequestMessageType, ParticipantResponseMessageType,
};
use crate::pseudonym_service_factory::{create_pseudonym_service, OauthAuth};
use actix_web::web::ServiceConfig;
use actix_web::{error::ErrorUnauthorized, middleware::Logger, web, HttpResponse, ResponseError};
use actix_web_httpauth::middleware::HttpAuthentication;
use log::error;
use oauth_token_service::TokenServiceConfig;
use paas_client::prelude::PAASConfig;
use paas_client::pseudonym_service::{PseudonymService, PseudonymServiceError};
use rand::rngs::OsRng;
use std::fmt::{Display, Formatter};
use std::future::Future;
use std::sync::Arc;
use tokio::sync::Mutex;

#[derive(Debug)]
pub enum ApiError {
    NotFound,
    Internal(String),
    BadRequest(String),
}

impl Display for ApiError {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        match self {
            ApiError::NotFound => write!(f, "Not found"),
            ApiError::Internal(msg) => write!(f, "Internal error: {}", msg),
            ApiError::BadRequest(msg) => write!(f, "Bad request: {}", msg),
        }
    }
}
impl ResponseError for ApiError {
    fn error_response(&self) -> HttpResponse {
        match self {
            ApiError::NotFound => HttpResponse::NotFound().json("Not found"),
            ApiError::Internal(msg) => HttpResponse::InternalServerError().json(msg),
            ApiError::BadRequest(msg) => HttpResponse::BadRequest().json(msg),
        }
    }
}
#[derive(Clone)]
pub struct DataSourceAPI {
    connector: Arc<dyn DataSource>,
    pseudonym_service: Arc<Mutex<PseudonymService>>,
    nest_auth_token: String,
}

impl DataSourceAPI {
    /// Create a new instance of the DataSourceAPI.
    pub async fn init(
        connector: impl DataSource,
        nest_auth_token: String,
        ps_config: PAASConfig,
        oauth_config: TokenServiceConfig,
    ) -> Self {
        let oauth_auth = OauthAuth::new(oauth_config);
        let pseudonym_service = create_pseudonym_service(oauth_auth, ps_config).await;
        Self {
            connector: Arc::new(connector),
            pseudonym_service,
            nest_auth_token,
        }
    }

    /// Configure an actix-web service with API routes and authentication middleware.
    pub fn configure_service(&self, cfg: &mut ServiceConfig) {
        let auth_token = self.nest_auth_token.clone();
        let connector = self.connector.clone();

        let auth = HttpAuthentication::bearer(move |req, credentials| {
            let auth_token = auth_token.clone();
            async move {
                if credentials.token() == auth_token {
                    Ok(req)
                } else {
                    Err((ErrorUnauthorized("Invalid token"), req))
                }
            }
        });

        cfg.app_data(web::Data::new(connector)).service(
            web::scope("/api")
                .app_data(web::Data::new(self.pseudonym_service.clone()))
                .wrap(auth)
                .route(
                    "/participant/create",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ParticipantRequestMessageType,
                                ParticipantResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.create_local_participant(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/create_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ParticipantBatchRequestMessageType,
                                ParticipantBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.create_local_participant_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/update",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ParticipantRequestMessageType,
                                ParticipantResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.update_local_participant(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/update_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ParticipantBatchRequestMessageType,
                                ParticipantBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.update_local_participant_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/delete",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ParticipantRequestMessageType,
                                ParticipantResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.delete_local_participant(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/delete_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ParticipantBatchRequestMessageType,
                                ParticipantBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.delete_local_participant_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/details",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ParticipantRequestMessageType,
                                ParticipantResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.local_participant_details(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/details_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ParticipantBatchRequestMessageType,
                                ParticipantBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.local_participant_details_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/create",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ActivityRequestMessageType,
                                ActivityResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.create_participant_activity(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/create_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ActivityBatchRequestMessageType,
                                ActivityBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.create_participant_activity_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/update",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ActivityRequestMessageType,
                                ActivityResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.update_participant_activity(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/update_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ActivityBatchRequestMessageType,
                                ActivityBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.update_participant_activity_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/delete",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ActivityRequestMessageType,
                                ActivityResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.delete_participant_activity(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/delete_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ActivityBatchRequestMessageType,
                                ActivityBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.delete_participant_activity_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/details",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ActivityRequestMessageType,
                                ActivityResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.participant_activity_details(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/details_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ActivityBatchRequestMessageType,
                                ActivityBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.participant_activity_details_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/progress",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ActivityRequestMessageType,
                                ActivityResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.participant_activity_progress(r), req, ps
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/progress_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ActivityBatchRequestMessageType,
                                ActivityBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.participant_activity_progress_batch(r),
                                req,
                                ps,
                            )
                            .await
                        }
                    }),
                )
                .route(
                    "/participant/activities/result",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep::<
                                ActivityRequestMessageType,
                                ActivityResponseMessageType,
                                _,
                                _,
                            >(
                                |r| connector.participant_activity_result(r), req, ps
                            )
                            .await
                        }
                    }), // TODO: maybe we should require different authentication for this route, since we only want end-users to access it
                )
                .route(
                    "/participant/activities/result_batch",
                    web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
                        let connector = connector.clone();
                        async move {
                            handle_pep_batch::<
                                ActivityBatchRequestMessageType,
                                ActivityBatchResponseMessageType,
                                PEPPseudoDataPoint,
                                PseudoDataPoint,
                                _,
                                _,
                            >(
                                |r| connector.participant_activity_result_batch(r), req, ps
                            )
                            .await
                        }
                    }),
                ),
        );
        // .route("/participants/create", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.create_local_participants(r), req, ps).await }
        // }))
        // .route("/participants/update", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.update_participants(r), req, ps).await }
        // }))
        // .route("/participants/delete", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.delete_participants(r), req, ps).await }
        // }))
        // .route("/participants/details", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.delete_participants(r), req, ps).await }
        // }))
        // .route("/participants/activities/create", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.create_participants_activity(r), req, ps).await }
        // }))
        // .route("/participants/activities/update", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.update_participants_activity(r), req, ps).await }
        // }))
        // .route("/participants/activities/delete", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.delete_participants_activity(r), req, ps).await }
        // }))
        // .route("/participants/activities/details", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.participants_activity_details(r), req, ps).await }
        // }))
        // .route("/participants/activities/progress", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.participants_activity_progress(r), req, ps).await }
        // }))
        // .route("/participants/activities/result", web::post().to(|connector: web::Data<Arc<dyn DataSource>>, req, ps| {
        //     let connector = connector.clone();
        //     async move { handle(|r| connector.participants_activity_result(r), req, ps).await }
        // }))
    }

    /// Run as a standalone server (if needed)
    pub async fn run_standalone(self, host: &str, port: u16) -> std::io::Result<()> {
        use actix_web::{App, HttpServer};

        HttpServer::new(move || {
            App::new()
                .wrap(Logger::default())
                .configure(|cfg| self.configure_service(cfg))
        })
        .bind((host, port))?
        .run()
        .await
    }
}

async fn handle_pep<Req, Res, F, Fut>(
    handle: F,
    request: web::Json<Req::PEPMessage>,
    pseudonym_service: web::Data<Arc<Mutex<PseudonymService>>>,
) -> Result<HttpResponse, ApiError>
where
    Req: PEPMessageType,            // Request message type
    Res: PEPMessageType,            // Response message type
    F: FnOnce(Req::Message) -> Fut, // Async handler type
    Fut: Future<Output = Result<Option<Res::Message>, ApiError>>, // Handler output
{
    let rng = &mut OsRng;
    let pep_request = request.into_inner();
    let domain_to = pep_request.domain_to().clone();

    let mut ps = pseudonym_service.lock().await;

    let request = Req::unpack(pep_request, &mut ps).await?;
    let response: Option<Res::Message> = handle(request).await?;

    if let Some(response) = response {
        let pep_response = Res::pack(response, domain_to, &mut ps, rng).await?;
        Ok(HttpResponse::Ok().json(pep_response))
    } else {
        Ok(HttpResponse::Ok().finish())
    }
}

async fn handle_pep_batch<Req, Res, PepT, T, F, Fut>(
    handle: F,
    request: web::Json<Req::PEPBatchMessage>,
    pseudonym_service: web::Data<Arc<Mutex<PseudonymService>>>,
) -> Result<HttpResponse, ApiError>
where
    Req: PEPBatchMessageType<PepT, T>,   // Request message type
    Res: PEPBatchMessageType<PepT, T>,   // Response message type
    F: FnOnce(Req::BatchMessage) -> Fut, // Async handler type
    Fut: Future<Output = Result<Option<Res::BatchMessage>, ApiError>>, // Handler output
{
    let rng = &mut OsRng;
    let pep_request = request.into_inner();

    let domain_to = pep_request.domain_to().clone();

    let mut ps = pseudonym_service.lock().await;

    let request = Req::unpack(pep_request, &mut ps).await?;
    let response: Option<Res::BatchMessage> = handle(request).await?;

    if let Some(response) = response {
        let pep_response = Res::pack(response, domain_to, &mut ps, rng).await?;
        Ok(HttpResponse::Ok().json(pep_response))
    } else {
        Ok(HttpResponse::Ok().finish())
    }
}

pub(crate) trait PseudonymServiceErrorHandler<T> {
    fn handle_pseudonym_error(self) -> Result<T, ApiError>;
}

impl<T> PseudonymServiceErrorHandler<T> for Result<T, PseudonymServiceError> {
    fn handle_pseudonym_error(self) -> Result<T, ApiError> {
        // TODO: Specific error handling
        self.map_err(|e| {
            error!("Pseudonym service error: {e:?}");
            Internal("Pseudonym service operation failed".to_string())
        })
    }
}