Getting started with MongoDB and Redis in Rust

pub async fn get_planet(&self, planet_id: &str) -> Result<Planet, CustomError> {
    let cache_key = self.get_planet_cache_key(planet_id);
    let mut con = self.redis_client.get_async_connection().await?;

    let cached_planet = con.get(&cache_key).await?;
    match cached_planet {
        Value::Nil => {
            debug!("Use database to retrieve a planet by id: {}", &planet_id);
            let result: Planet = self
                .mongodb_client
                .get_planet(ObjectId::from_str(planet_id)?)
                .await?;

            let _: () = redis::pipe()
                .atomic()
                .set(&cache_key, &result)
                .expire(&cache_key, 60)
                .query_async(&mut con)
                .await?;

            Ok(result)
        }
        Value::Data(val) => {
            debug!("Use cache to retrieve a planet by id: {}", planet_id);
            Ok(serde_json::from_slice(&val)?)
        }
        _ => Err(RedisError {
            message: "Unexpected response from Redis".to_string(),
        }),
    }
}

impl ToRedisArgs for &Planet {
    fn write_redis_args<W>(&self, out: &mut W)
    where
        W: ?Sized + RedisWrite,
    {
        out.write_arg_fmt(serde_json::to_string(self).expect("Can't serialize Planet as string"))
    }
}

pub async fn update_planet(
    &self,
    planet_id: &str,
    planet: Planet,
) -> Result<Planet, CustomError> {
    let updated_planet = self
        .mongodb_client
        .update_planet(ObjectId::from_str(planet_id)?, planet)
        .await?;

    let cache_key = self.get_planet_cache_key(planet_id);
    self.redis_connection_manager.clone().del(cache_key).await?;

    Ok(updated_planet)
}

pub async fn get_image_of_planet(&self, planet_id: &str) -> Result<Vec<u8>, CustomError> {
    let cache_key = self.get_image_cache_key(planet_id);
    let mut redis_connection_manager = self.redis_connection_manager.clone();

    let cached_image = redis_connection_manager.get(&cache_key).await?;
    match cached_image {
        Value::Nil => {
            debug!(
                "Use database to retrieve an image of a planet by id: {}",
                &planet_id
            );
            let planet = self
                .mongodb_client
                .get_planet(ObjectId::from_str(planet_id)?)
                .await?;
            let result = crate::db::get_image_of_planet(&planet.name).await;

            let _: () = redis::pipe()
                .atomic()
                .set(&cache_key, result.clone())
                .expire(&cache_key, 60)
                .query_async(&mut redis_connection_manager)
                .await?;

            Ok(result)
        }
        Value::Data(val) => {
            debug!(
                "Use cache to retrieve an image of a planet by id: {}",
                &planet_id
            );
            Ok(val)
        }
        _ => Err(RedisError {
            message: "Unexpected response from Redis".to_string(),
        }),
    }
}

#[derive(Clone)]
pub struct RateLimitingService {
    redis_connection_manager: ConnectionManager,
}

impl RateLimitingService {
    pub fn new(redis_connection_manager: ConnectionManager) -> Self {
        RateLimitingService {
            redis_connection_manager,
        }
    }

    pub async fn assert_rate_limit_not_exceeded(&self, ip_addr: String) -> Result<(), CustomError> {
        let current_minute = Utc::now().minute();
        let rate_limit_key = format!("{}:{}:{}", RATE_LIMIT_KEY_PREFIX, ip_addr, current_minute);

        let (count,): (u64,) = redis::pipe()
            .atomic()
            .incr(&rate_limit_key, 1)
            .expire(&rate_limit_key, 60)
            .ignore()
            .query_async(&mut self.redis_connection_manager.clone())
            .await?;

        if count > MAX_REQUESTS_PER_MINUTE {
            Err(TooManyRequests {
                actual_count: count,
                permitted_count: MAX_REQUESTS_PER_MINUTE,
            })
        } else {
            Ok(())
        }
    }
}

pub async fn get_planets(
    req: HttpRequest,
    web::Query(query_params): web::Query<GetPlanetsQueryParams>,
    rate_limit_service: web::Data<Arc<RateLimitingService>>,
    planet_service: web::Data<Arc<PlanetService>>,
) -> Result<HttpResponse, CustomError> {
    rate_limit_service
        .assert_rate_limit_not_exceeded(get_ip_addr(&req)?)
        .await?;

    let planets = planet_service.get_planets(query_params.r#type).await?;
    Ok(HttpResponse::Ok().json(planets.into_iter().map(PlanetDto::from).collect::<Vec<_>>()))
}

pub async fn create_planet(&self, planet: Planet) -> Result<Planet, CustomError> {
    let planet = self.mongodb_client.create_planet(planet).await?;
    self.redis_connection_manager
        .clone()
        .publish(
            NEW_PLANETS_CHANNEL_NAME,
            serde_json::to_string(&PlanetMessage::from(&planet))?,
        )
        .await?;
    Ok(planet)
}

pub async fn get_new_planets_stream(
    &self,
) -> Result<Receiver<Result<Bytes, CustomError>>, CustomError> {
    let (tx, rx) = mpsc::channel::<Result<Bytes, CustomError>>(100);

    tx.send(Ok(Bytes::from("data: Connected\n\n")))
        .await
        .expect("Can't send a message to the stream");

    let mut pubsub_con = self
        .redis_client
        .get_async_connection()
        .await?
        .into_pubsub();
    pubsub_con.subscribe(NEW_PLANETS_CHANNEL_NAME).await?;

    tokio::spawn(async move {
        while let Some(msg) = pubsub_con.on_message().next().await {
            let payload = msg.get_payload().expect("Can't get payload of message");
            let payload: String = FromRedisValue::from_redis_value(&payload)
                .expect("Can't convert from Redis value");
            let msg = Bytes::from(format!("data: Planet created: {:?}\n\n", payload));
            tx.send(Ok(msg))
                .await
                .expect("Can't send a message to the stream");
        }
    });

    Ok(rx)
}

pub async fn sse(
    planet_service: web::Data<Arc<PlanetService>>,
) -> Result<HttpResponse, CustomError> {
    let new_planets_stream = planet_service.get_new_planets_stream().await?;
    let response_stream = tokio_stream::wrappers::ReceiverStream::new(new_planets_stream);

    Ok(HttpResponse::build(StatusCode::OK)
        .insert_header(header::ContentType(mime::TEXT_EVENT_STREAM))
        .streaming(response_stream))
}

curl -X POST -H 'Content-Type: application/json' -d '{
    \"name\": \"Pluto\",
    \"type\": \"DwarfPlanet\",
    \"mean_radius\": 1188,
    \"satellites\": null
}' localhost:9000/planets

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    ...

    let enable_write_handlers = env::var("ENABLE_WRITE_HANDLERS")
        .expect("ENABLE_WRITE_HANDLERS env var should be specified")
        .parse::<bool>()
        .expect("Can't parse ENABLE_WRITE_HANDLERS");

    HttpServer::new(move || {
        let mut app = App::new()
            .route("/planets", web::get().to(handlers::get_planets))
            .route("/planets/{planet_id}", web::get().to(handlers::get_planet))
            .route(
                "/planets/{planet_id}/image",
                web::get().to(handlers::get_image_of_planet),
            )
            .route("/events", web::get().to(handlers::sse))
            .route("/", web::get().to(handlers::index))
            .data(Arc::clone(&planet_service))
            .data(Arc::clone(&rate_limiting_service));

        if enable_write_handlers {
            app = app
                .route("/planets", web::post().to(handlers::create_planet))
                .route(
                    "/planets/{planet_id}",
                    web::put().to(handlers::update_planet),
                )
                .route(
                    "/planets/{planet_id}",
                    web::delete().to(handlers::delete_planet),
                );
        }

        app
    })
    .bind("0.0.0.0:9000")?
    .run()
    .await
}