GKaszewski/rick-and-morty-tournament
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Rewrite from mongodb to sqlx (#1)

Browse Source 
Reviewed-on: #1
This commit is contained in:
Gabriel Kaszewski
2025-07-20 13:16:31 +00:00
parent a58df1cb8e
commit 14c520c1d2
19 changed files with 814 additions and 1078 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
src/bin/fetch_characters.rs
View File

@@ -1,11 +1,7 @@
use mongodb::{
Client,
bson::{doc, to_document},
options::UpdateOptions,
};
use reqwest::Client as HttpClient;
use rick_and_morty::models::{Character, OriginOrLocation};
use serde::{Deserialize, Serialize};
use sqlx::{Row, SqlitePool};
#[derive(Debug, Deserialize, Serialize)]
struct ApiCharacter {
@@ -27,17 +23,20 @@ struct ApiCharacter {
impl From<ApiCharacter> for Character {
fn from(api: ApiCharacter) -> Self {
Character {
id: None, // always None for new/incoming data
id: 0, // Ignored for new/incoming data; SQLite autoincrements
rmid: api.id,
name: api.name,
status: api.status,
species: api.species,
r#type: api.character_type,
gender: api.gender,
origin: api.origin,
location: api.location,
origin_name: api.origin.name,
origin_url: api.origin.url,
location_name: api.location.name,
location_url: api.location.url,
image: api.image,
episode: api.episode,
// Store as JSON string
episode: serde_json::to_string(&api.episode).unwrap(),
url: api.url,
created: api.created,
elo_rating: 1000.0,
@@ -58,13 +57,9 @@ fn init_tracing() {
async fn main() -> anyhow::Result<()> {
init_tracing();
dotenvy::dotenv().ok();
let db_uri =
std::env::var("MONGODB_URI").unwrap_or_else(|_| "mongodb://localhost:27017".to_string());
let db_name = std::env::var("DB_NAME").unwrap_or_else(|_| "rick_and_morty".to_string());
let client = Client::with_uri_str(&db_uri).await?;
let db = client.database(&db_name);
let collection = db.collection::<Character>("characters");
let db_url = std::env::var("DATABASE_URL").expect("DATABASE_URL not set");
let pool = SqlitePool::connect(&db_url).await?;
tracing::info!("Starting to fetch characters from Rick and Morty API");
let http = HttpClient::new();
@@ -92,30 +87,77 @@ async fn main() -> anyhow::Result<()> {
count = all_characters.len(),
"Fetched all characters, starting DB upsert"
);
let options = UpdateOptions::builder().upsert(true).build();
// let insert_result = collection.insert_many(all_characters.clone()).await?;
for character in &all_characters {
let filter = doc! { "rmid": character.rmid };
let mut set_doc = to_document(character)?;
set_doc.remove("elo_rating"); // Do NOT overwrite existing Elo
let update = doc! {
"$set": set_doc,
"$setOnInsert": { "elo_rating": 1000.0 }
};
if let Err(e) = collection
.update_one(filter, update)
.with_options(Some(options.clone()))
.await
{
tracing::error!(error = ?e, id = character.rmid, name = %character.name, "Failed to upsert character");
// Try to fetch existing character by rmid
let row = sqlx::query("SELECT id, elo_rating FROM characters WHERE rmid = ?")
.bind(character.rmid)
.fetch_optional(&pool)
.await?;
if let Some(row) = row {
// Exists: update all fields except elo_rating
let id: i64 = row.get("id");
sqlx::query(
"UPDATE characters SET
name = ?, status = ?, species = ?, type = ?, gender = ?,
origin_name = ?, origin_url = ?,
location_name = ?, location_url = ?,
image = ?, episode = ?, url = ?, created = ?
WHERE id = ?",
)
.bind(&character.name)
.bind(&character.status)
.bind(&character.species)
.bind(&character.r#type)
.bind(&character.gender)
.bind(&character.origin_name)
.bind(&character.origin_url)
.bind(&character.location_name)
.bind(&character.location_url)
.bind(&character.image)
.bind(&character.episode)
.bind(&character.url)
.bind(&character.created)
.bind(id)
.execute(&pool)
.await?;
tracing::info!(id = character.rmid, name = %character.name, "Updated character (no elo overwrite)");
} else {
// Insert new character with default elo_rating
sqlx::query(
"INSERT INTO characters (
rmid, name, status, species, type, gender,
origin_name, origin_url, location_name, location_url,
image, episode, url, created, elo_rating
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
)
.bind(character.rmid)
.bind(&character.name)
.bind(&character.status)
.bind(&character.species)
.bind(&character.r#type)
.bind(&character.gender)
.bind(&character.origin_name)
.bind(&character.origin_url)
.bind(&character.location_name)
.bind(&character.location_url)
.bind(&character.image)
.bind(&character.episode)
.bind(&character.url)
.bind(&character.created)
.bind(1000.0)
.execute(&pool)
.await?;
tracing::info!(id = character.rmid, name = %character.name, "Inserted new character");
}
tracing::info!(id = character.rmid, name = %character.name, "Upserted character");
}
// tracing::info!("Inserted {} characters", insert_result.inserted_ids.len());
let character_count = collection.count_documents(doc! {}).await?;
let character_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM characters")
.fetch_one(&pool)
.await?;
tracing::info!(
count = character_count,
"Total characters in DB after import"

13
src/db.rs
View File

@@ -1,11 +1,6 @@
use mongodb::{Client, Database};
use once_cell::sync::OnceCell;
use sqlx::sqlite::SqlitePool;
pub static DB: OnceCell<Database> = OnceCell::new();
pub async fn connect_db(uri: &str, db_name: &str) -> mongodb::error::Result<()> {
let client = Client::with_uri_str(uri).await?;
let db = client.database(db_name);
DB.set(db).ok();
Ok(())
pub async fn connect_db(db_url: &str) -> Result<SqlitePool, sqlx::Error> {
let pool = SqlitePool::connect(db_url).await?;
Ok(pool)
}

20
src/main.rs
View File

@@ -20,19 +20,17 @@ fn init_tracing() {
}
#[tokio::main]
async fn main() {
async fn main() -> anyhow::Result<()> {
init_tracing();
dotenvy::dotenv().ok();
let db_uri = std::env::var("MONGODB_URI").expect("MONGODB_URI not set");
let db_name = std::env::var("DB_NAME").unwrap_or_else(|_| "rick_and_morty".to_string());
let address = std::env::var("BIND_ADDR").unwrap_or_else(|_| "0.0.0.0:8000".to_string());
let db_url = std::env::var("DATABASE_URL").expect("DATABASE_URL not set");
db::connect_db(&db_uri, &db_name)
.await
.expect("Failed to connect to database");
let db = db::connect_db(&db_url).await?;
let db = db::DB.get().expect("Database not initialized");
// migrations
sqlx::migrate!().run(&db).await?;
let cors = CorsLayer::new()
// allow `GET` and `POST` when accessing the resource
@@ -44,7 +42,7 @@ async fn main() {
let app = Router::new()
.route("/characters", get(routes::get_characters))
.route("/rate", post(routes::rate))
.with_state(db.clone())
.with_state(db)
.layer(TraceLayer::new_for_http())
.layer(cors)
.fallback_service(ServeDir::new("frontend/dist").append_index_html_on_directories(true));
@@ -55,5 +53,9 @@ async fn main() {
println!("Listening on {}", listener.local_addr().unwrap());
axum::serve(listener, app).await.unwrap();
axum::serve(listener, app).await?;
tracing::info!("Server started successfully");
Ok(())
}

17
src/models/mod.rs
View File

@@ -1,26 +1,27 @@
use mongodb::bson::{doc, oid::ObjectId};
use serde::{Deserialize, Serialize};
use sqlx::prelude::FromRow;
#[derive(Debug, Serialize, Deserialize, Clone)]
#[derive(Debug, Serialize, Deserialize, Clone, FromRow)]
pub struct OriginOrLocation {
pub name: String,
pub url: String,
}
#[derive(Debug, Serialize, Deserialize, Clone)]
#[derive(Debug, Serialize, Deserialize, Clone, FromRow)]
pub struct Character {
#[serde(rename = "_id", skip_serializing_if = "Option::is_none")]
pub id: Option<ObjectId>, // Mongo _id
pub id: i64,
pub rmid: i32, // Rick&Morty ID, don't confuse with _id
pub name: String,
pub status: String,
pub species: String,
pub r#type: String,
pub gender: String,
pub origin: OriginOrLocation,
pub location: OriginOrLocation,
pub origin_name: String,
pub origin_url: String,
pub location_name: String,
pub location_url: String,
pub image: String,
pub episode: Vec<String>,
pub episode: String,
pub url: String,
pub created: String,
pub elo_rating: f64,

113
src/routes/mod.rs
View File

@@ -1,10 +1,7 @@
use axum::response::Html;
use axum::{Json, extract::State, http::StatusCode};
use futures_util::stream::TryStreamExt;
use mongodb::Database;
use mongodb::bson::doc;
use mongodb::bson::oid::ObjectId;
use serde::{Deserialize, Serialize};
use sqlx::{Sqlite, SqlitePool, Transaction};
use crate::models::Character;
use crate::utils::calculate_elo;
@@ -15,7 +12,7 @@ pub struct RateRequest {
loser_id: String,
}
#[derive(Deserialize, Serialize)]
#[derive(Serialize)]
pub struct RateResponse {
winner: Character,
loser: Character,
@@ -24,94 +21,74 @@ pub struct RateResponse {
static K_FACTOR: f64 = 32.0; // K-factor for Elo rating system
pub async fn get_characters(
State(db): State<Database>,
State(db): State<SqlitePool>,
) -> Result<Json<Vec<Character>>, StatusCode> {
let collection = db.collection::<Character>("characters");
let pipeline = vec![doc! { "$sort": { "elo_rating": -1 } }];
let mut cursor = collection
.aggregate(pipeline)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let mut characters = Vec::new();
while let Some(result) = cursor
.try_next()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
{
let character: Character =
mongodb::bson::from_document(result).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
characters.push(character);
}
tracing::info!(count = characters.len(), "Fetched characters from DB");
let characters =
sqlx::query_as::<_, Character>("SELECT * FROM characters ORDER BY elo_rating DESC")
.fetch_all(&db)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(characters))
}
#[axum::debug_handler]
pub async fn rate(
State(db): State<Database>,
State(db): State<SqlitePool>,
Json(req): Json<RateRequest>,
) -> Result<Json<RateResponse>, StatusCode> {
let collection = db.collection::<Character>("characters");
let winner_oid = ObjectId::parse_str(&req.winner_id).map_err(|_| StatusCode::BAD_REQUEST)?;
let loser_oid = ObjectId::parse_str(&req.loser_id).map_err(|_| StatusCode::BAD_REQUEST)?;
let winner = collection
.find_one(doc! { "_id": winner_oid })
// 1. Fetch characters by ID
let winner: Character = sqlx::query_as("SELECT * FROM characters WHERE id = ?")
.bind(&req.winner_id)
.fetch_optional(&db)
.await
.ok()
.flatten()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
let loser = collection
.find_one(doc! { "_id": loser_oid })
let loser: Character = sqlx::query_as("SELECT * FROM characters WHERE id = ?")
.bind(&req.loser_id)
.fetch_optional(&db)
.await
.ok()
.flatten()
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
let (new_winner_elo, new_loser_elo) =
calculate_elo(winner.elo_rating, loser.elo_rating, K_FACTOR);
collection
.update_one(
doc! {"_id": &winner_oid},
doc! { "$set": { "elo_rating": new_winner_elo } },
)
let mut tx: Transaction<'_, Sqlite> = db
.begin()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
collection
.update_one(
doc! {"_id": &loser_oid},
doc! { "$set": { "elo_rating": new_loser_elo } },
)
sqlx::query("UPDATE characters SET elo_rating = ? WHERE id = ?")
.bind(new_winner_elo)
.bind(&req.winner_id)
.execute(&mut *tx)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
// return resposnse with characters' new Elo ratings
let updated_winner = collection
.find_one(doc! { "_id": winner_oid })
sqlx::query("UPDATE characters SET elo_rating = ? WHERE id = ?")
.bind(new_loser_elo)
.bind(&req.loser_id)
.execute(&mut *tx)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
let updated_loser = collection
.find_one(doc! { "_id": loser_oid })
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?
.ok_or(StatusCode::NOT_FOUND)?;
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
tracing::info!(
"Rated characters: winner_id = {}, loser_id = {}, new_winner_elo =
{}, new_loser_elo = {}",
req.winner_id,
req.loser_id,
new_winner_elo,
new_loser_elo
);
tx.commit()
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let updated_winner: Character = sqlx::query_as("SELECT * FROM characters WHERE id = ?")
.bind(&req.winner_id)
.fetch_one(&db)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
let updated_loser: Character = sqlx::query_as("SELECT * FROM characters WHERE id = ?")
.bind(&req.loser_id)
.fetch_one(&db)
.await
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(Json(RateResponse {
winner: updated_winner,