Building a trait service with Rust and Kafka

#[macro_use]
extern crate rocket;

// Load modules
mod event;
mod kafka;

use event::user::controllers::user_controller;

#[launch]
fn rocket() -> _ {
    rocket::build().mount("/", routes![user_controller::track])
}

// Load test files
#[cfg(test)]
mod tests;

// # Goals of the controller files

// ## Define a route
// The controller establishes a connection between a specific URL path (route) and the corresponding function that handles the request.
// This tells the application how to direct incoming requests to the appropriate logic within the controller.

// ## Parse the incoming request payload
// The request payload contains data sent by the client (like form data or JSON objects).
// The controller parses this data into a usable format, making it accessible for further processing.

// ## Validate the payload
// It's crucial to ensure the received data is in the expected format and adheres to any defined rules.
// The controller validates the payload to prevent unexpected behavior or security vulnerabilities.

// ## Communicate with other services
// Controllers often interact with other parts of the application, like models or services, to retrieve or manipulate data.
// This communication allows the controller to fulfill the request's purpose.

// ## Return a response
// After processing the request, the controller generates a response object containing the data or message to be sent back to the client.
// This response could be HTML content, JSON data, or any format appropriate for the request type.

use rocket::serde::json::{serde_json, Json};

use crate::event::user::models::event::UserEvent;
use crate::kafka;

// Track user interactions with the product
// Send event to the kafka server and return 200 http code
// Message consumers will process the event asynchronously
// Failure points
// - Kafka server is down
// - Invalid event payload
#[post("/event/user/track", format = "json", data = "<event>")]
pub async fn track(event: Json<UserEvent<'_>>) -> &'static str {
    // TODO: Validate event payload
    let payload = serde_json::to_string(&event.payload).unwrap();

    // Send event to the kafka server
    kafka::send_event(event.key, &payload).await;

    "ok"
}

// # Goals of the model files

// ## Define data structures
// This ensures consistency in how data is stored and accessed throughout your application.

// ## Show model examples
// This provides practical demonstrations of how the model is used, making it easier for developers to understand and implement.

// ## Define enum values
// This helps limit data to specific options and reduces the chance of errors or unexpected values being entered.

// ## Reduce ambiguity
// By clearly defining the model's structure and behavior, you minimize confusion and make the code more maintainable.

use rocket::serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize, Debug)]
#[serde(crate = "rocket::serde")]
pub struct UserEventPayload<'a> {
    pub client_os: Option<&'a str>,
}

// UserEvent captures the interaction of a user with the product
// the key should be unique and self-explanatory.
// It should respond to who, what, where, when, why & how
// key: v1.landing.user.order.create
// payload: { client_os: "iOS", client_version: "1.0.0" }
#[derive(Serialize, Deserialize, Debug)]
#[serde(crate = "rocket::serde")]
pub struct UserEvent<'a> {
    pub key: &'a str,
    pub payload: UserEventPayload<'a>,
}

use std::time::Duration;

use rdkafka::config::ClientConfig;
use rdkafka::producer::{FutureProducer, FutureRecord};

// TODO: Move constants
const KAFKA_TOPIC_NAME: &str = "user_events";

// Send event to the kafka server and return the delivery status
pub async fn send_event(key: &str, message: &str) {
    let producer: FutureProducer = kafka_producer().expect("Producer creation error");

    producer
        .send(
            FutureRecord::to(KAFKA_TOPIC_NAME).payload(message).key(key),
            Duration::from_secs(0),
        )
        .await;
}

// Create kafka producer
fn kafka_producer() -> Result<FutureProducer, rdkafka::error::KafkaError> {
    // Get the kafka broker url from the environment variable or use the default value
    let kafka_broker_url: String =
        std::env::var("KAFKA_BROKER_URL").unwrap_or("localhost:9093".to_string());

    ClientConfig::new()
        .set("bootstrap.servers", kafka_broker_url)
        .set("message.timeout.ms", "5000")
        .create()
}

use crate::rocket;
use rocket::http::ContentType;
use rocket::http::Status;
use rocket::local::blocking::{Client, LocalResponse};
use rocket::serde::json::{json, serde_json};

fn rocket_client() -> Client {
    Client::tracked(rocket()).expect("valid rocket instance")
}

fn event(key: &str, payload: serde_json::Value) -> String {
    json!({
        "key": key,
        "payload": payload,
    })
    .to_string()
}

fn empty_payload() -> serde_json::Value {
    json!({})
}

fn invalid_payload() -> serde_json::Value {
    json!({
        "client_os": 33,
    })
}

fn post_user_event<'a>(client: &'a Client, payload: &'a str) -> LocalResponse<'a> {
    return client
        .post("/event/user/track")
        .header(ContentType::JSON)
        .body(payload)
        .dispatch();
}

#[test]
fn track_empty_payload() {
    let client = rocket_client();
    let payload = event("landing.user.home.click", empty_payload());
    let response = post_user_event(&client, &payload);

    assert_eq!(response.status(), Status::Ok);
}

#[test]
fn track_invalid_payload() {
    let client = rocket_client();
    let payload = event("landing.user.home.click", invalid_payload());
    let response = post_user_event(&client, &payload);

    assert_eq!(response.status(), Status::UnprocessableEntity);
}