diff --git a/apps/http-server/src/api/admin/shows.rs b/apps/http-server/src/api/admin/shows.rs index 0d4795d..dd61719 100644 --- a/apps/http-server/src/api/admin/shows.rs +++ b/apps/http-server/src/api/admin/shows.rs @@ -268,9 +268,15 @@ pub async fn create_show( let mut show: Show = body.into(); show.created_at = Some(Utc::now()); - - let mut session = state.mongo_client.start_session().await.map_err(|e| AppError::internal(e.to_string()))?; - session.start_transaction().await.map_err(|e| AppError::internal(e.to_string()))?; + let mut session = state + .mongo_client + .start_session() + .await + .map_err(|e| AppError::internal(e.to_string()))?; + session + .start_transaction() + .await + .map_err(|e| AppError::internal(e.to_string()))?; let col = shows_col(&state); let result = col @@ -279,11 +285,14 @@ pub async fn create_show( .await .map_err(|e| AppError::internal(e.to_string()))?; - let lsn_col = state.mongo_client.database(&state.mongo_db_name).collection::("search_outbox_lsn"); + let lsn_col = state + .mongo_client + .database(&state.mongo_db_name) + .collection::("search_outbox_lsn"); let lsn_doc = lsn_col .find_one_and_update( doc! { "_id": "elasticsearch_outbox" }, - doc! { "$inc": { "sequence": 1_i64 } } + doc! { "$inc": { "sequence": 1_i64 } }, ) .upsert(true) .return_document(mongodb::options::ReturnDocument::After) @@ -295,7 +304,10 @@ pub async fn create_show( let sequence = lsn_doc.get_i64("sequence").unwrap_or(1); let inserted_id = result.inserted_id.as_object_id().unwrap_or_default(); - let outbox_col = state.mongo_client.database(&state.mongo_db_name).collection::("search_outbox_events"); + let outbox_col = state + .mongo_client + .database(&state.mongo_db_name) + .collection::("search_outbox_events"); let outbox_event = doc! { "sequence": sequence, "show_id": inserted_id.to_hex(), @@ -304,13 +316,19 @@ pub async fn create_show( "trace_context": bson::Bson::Null, "created_at": bson::DateTime::now(), }; - outbox_col.insert_one(outbox_event).session(&mut session).await.map_err(|e| AppError::internal(e.to_string()))?; + outbox_col + .insert_one(outbox_event) + .session(&mut session) + .await + .map_err(|e| AppError::internal(e.to_string()))?; - session.commit_transaction().await.map_err(|e| AppError::internal(e.to_string()))?; + session + .commit_transaction() + .await + .map_err(|e| AppError::internal(e.to_string()))?; let inserted_id = inserted_id.to_hex(); - invalidate_async(&state, bookit_redis::keys::CACHE_SHOWS).await; invalidate_async(&state, bookit_redis::keys::CACHE_DASHBOARD_GRID).await; @@ -342,20 +360,29 @@ pub async fn update_show( show.id = Some(oid); show.created_at = existing.created_at; - - let mut session = state.mongo_client.start_session().await.map_err(|e| AppError::internal(e.to_string()))?; - session.start_transaction().await.map_err(|e| AppError::internal(e.to_string()))?; + let mut session = state + .mongo_client + .start_session() + .await + .map_err(|e| AppError::internal(e.to_string()))?; + session + .start_transaction() + .await + .map_err(|e| AppError::internal(e.to_string()))?; col.replace_one(doc! { "_id": oid }, &show) .session(&mut session) .await .map_err(|e| AppError::internal(e.to_string()))?; - let lsn_col = state.mongo_client.database(&state.mongo_db_name).collection::("search_outbox_lsn"); + let lsn_col = state + .mongo_client + .database(&state.mongo_db_name) + .collection::("search_outbox_lsn"); let lsn_doc = lsn_col .find_one_and_update( doc! { "_id": "elasticsearch_outbox" }, - doc! { "$inc": { "sequence": 1_i64 } } + doc! { "$inc": { "sequence": 1_i64 } }, ) .upsert(true) .return_document(mongodb::options::ReturnDocument::After) @@ -365,7 +392,10 @@ pub async fn update_show( .ok_or_else(|| AppError::internal("Failed to generate LSN"))?; let sequence = lsn_doc.get_i64("sequence").unwrap_or(1); - let outbox_col = state.mongo_client.database(&state.mongo_db_name).collection::("search_outbox_events"); + let outbox_col = state + .mongo_client + .database(&state.mongo_db_name) + .collection::("search_outbox_events"); let outbox_event = doc! { "sequence": sequence, "show_id": oid.to_hex(), @@ -374,10 +404,16 @@ pub async fn update_show( "trace_context": bson::Bson::Null, "created_at": bson::DateTime::now(), }; - outbox_col.insert_one(outbox_event).session(&mut session).await.map_err(|e| AppError::internal(e.to_string()))?; - - session.commit_transaction().await.map_err(|e| AppError::internal(e.to_string()))?; + outbox_col + .insert_one(outbox_event) + .session(&mut session) + .await + .map_err(|e| AppError::internal(e.to_string()))?; + session + .commit_transaction() + .await + .map_err(|e| AppError::internal(e.to_string()))?; invalidate_async(&state, &cache_show_key(&id)).await; invalidate_async(&state, &cache_schedules_active_key()).await; @@ -396,9 +432,16 @@ pub async fn delete_show( let oid = ObjectId::parse_str(&id).map_err(|_| AppError::bad_request("Invalid show id"))?; let col = shows_col(&state); - - let mut session = state.mongo_client.start_session().await.map_err(|e| AppError::internal(e.to_string()))?; - session.start_transaction().await.map_err(|e| AppError::internal(e.to_string()))?; + + let mut session = state + .mongo_client + .start_session() + .await + .map_err(|e| AppError::internal(e.to_string()))?; + session + .start_transaction() + .await + .map_err(|e| AppError::internal(e.to_string()))?; col.update_one( doc! { "_id": oid }, @@ -408,11 +451,14 @@ pub async fn delete_show( .await .map_err(|e| AppError::internal(e.to_string()))?; - let lsn_col = state.mongo_client.database(&state.mongo_db_name).collection::("search_outbox_lsn"); + let lsn_col = state + .mongo_client + .database(&state.mongo_db_name) + .collection::("search_outbox_lsn"); let lsn_doc = lsn_col .find_one_and_update( doc! { "_id": "elasticsearch_outbox" }, - doc! { "$inc": { "sequence": 1_i64 } } + doc! { "$inc": { "sequence": 1_i64 } }, ) .upsert(true) .return_document(mongodb::options::ReturnDocument::After) @@ -422,7 +468,10 @@ pub async fn delete_show( .ok_or_else(|| AppError::internal("Failed to generate LSN"))?; let sequence = lsn_doc.get_i64("sequence").unwrap_or(1); - let outbox_col = state.mongo_client.database(&state.mongo_db_name).collection::("search_outbox_events"); + let outbox_col = state + .mongo_client + .database(&state.mongo_db_name) + .collection::("search_outbox_events"); let outbox_event = doc! { "sequence": sequence, "show_id": oid.to_hex(), @@ -431,10 +480,16 @@ pub async fn delete_show( "trace_context": bson::Bson::Null, "created_at": bson::DateTime::now(), }; - outbox_col.insert_one(outbox_event).session(&mut session).await.map_err(|e| AppError::internal(e.to_string()))?; - - session.commit_transaction().await.map_err(|e| AppError::internal(e.to_string()))?; + outbox_col + .insert_one(outbox_event) + .session(&mut session) + .await + .map_err(|e| AppError::internal(e.to_string()))?; + session + .commit_transaction() + .await + .map_err(|e| AppError::internal(e.to_string()))?; invalidate_async(&state, &cache_show_key(&id)).await; invalidate_async(&state, &cache_schedules_active_key()).await; diff --git a/apps/http-server/src/test_mongo.rs b/apps/http-server/src/test_mongo.rs index 439337e..7e1415a 100644 --- a/apps/http-server/src/test_mongo.rs +++ b/apps/http-server/src/test_mongo.rs @@ -1,12 +1,14 @@ -use mongodb::{Client, bson::doc}; use anyhow::Result; +use mongodb::{Client, bson::doc}; pub async fn test_tx(client: Client) -> Result<()> { let mut session = client.start_session().await?; session.start_transaction().await?; - let col = client.database("test").collection::("test"); - col.insert_one(doc!{"a": 1}).session(&mut session).await?; - + let col = client + .database("test") + .collection::("test"); + col.insert_one(doc! {"a": 1}).session(&mut session).await?; + session.commit_transaction().await?; Ok(()) } diff --git a/apps/search-server/src/outbox.rs b/apps/search-server/src/outbox.rs index e01a08b..5a29416 100644 --- a/apps/search-server/src/outbox.rs +++ b/apps/search-server/src/outbox.rs @@ -36,7 +36,7 @@ pub async fn ensure_collections(database: &Database) -> Result<()> { .options( IndexOptions::builder() .expire_after(std::time::Duration::from_secs(7 * 24 * 60 * 60)) - .build() + .build(), ) .build(), ) @@ -51,7 +51,7 @@ pub async fn claim_next(database: &Database) -> Result Result { .to_owned(), document: match value("document") { Bson::Null => None, - val => { - Some(mongodb::bson::from_bson(val).context("invalid search document payload")?) - } + val => Some(mongodb::bson::from_bson(val).context("invalid search document payload")?), }, trace_context: mongodb::bson::from_bson(value("trace_context")) .unwrap_or_else(|_| Value::Null),