fix: harden favorite state consistency

apps/api/src/modules/favorite/repo.rs

@@ -65,49 +65,31 @@ pub async fn upsert_product(
     user_id: Uuid,
     product_id: Uuid,
 ) -> ApiResult<Uuid> {
-    let inserted: Option<Uuid> = sqlx::query_scalar(
+    Ok(sqlx::query_scalar(
         "INSERT INTO favorites (user_id, product_id)
          VALUES ($1, $2)
          ON CONFLICT (user_id, product_id) WHERE product_id IS NOT NULL
-         DO NOTHING
+         DO UPDATE SET created_at = favorites.created_at
          RETURNING id",
     )
     .bind(user_id)
     .bind(product_id)
-    .fetch_optional(&mut *tx)
+    .fetch_one(&mut *tx)
-    .await?;
+    .await?)
-    if let Some(id) = inserted {
-        return Ok(id);
-    }
-    sqlx::query_scalar("SELECT id FROM favorites WHERE user_id = $1 AND product_id = $2")
-        .bind(user_id)
-        .bind(product_id)
-        .fetch_optional(&mut *tx)
-        .await?
-        .ok_or_else(|| ApiError::NotFound("favorite".into()))
 }
 
 pub async fn upsert_shop(tx: &mut PgConnection, user_id: Uuid, shop_id: Uuid) -> ApiResult<Uuid> {
-    let inserted: Option<Uuid> = sqlx::query_scalar(
+    Ok(sqlx::query_scalar(
         "INSERT INTO favorites (user_id, shop_id)
          VALUES ($1, $2)
          ON CONFLICT (user_id, shop_id) WHERE shop_id IS NOT NULL
-         DO NOTHING
+         DO UPDATE SET created_at = favorites.created_at
          RETURNING id",
     )
     .bind(user_id)
     .bind(shop_id)
-    .fetch_optional(&mut *tx)
+    .fetch_one(&mut *tx)
-    .await?;
+    .await?)
-    if let Some(id) = inserted {
-        return Ok(id);
-    }
-    sqlx::query_scalar("SELECT id FROM favorites WHERE user_id = $1 AND shop_id = $2")
-        .bind(user_id)
-        .bind(shop_id)
-        .fetch_optional(&mut *tx)
-        .await?
-        .ok_or_else(|| ApiError::NotFound("favorite".into()))
 }
 
 pub async fn delete_product(db: &PgPool, user_id: Uuid, product_id: Uuid) -> ApiResult<()> {

apps/api/tests/favorites.rs

@@ -462,3 +462,97 @@ async fn add_holds_target_visible_until_favorite_commits() {
     );
     assert_eq!(suspend_response.unwrap().status(), 200);
 }
+
+#[tokio::test]
+#[serial]
+async fn repeated_add_holds_existing_favorite_until_response_is_hydrated() {
+    let app = spawn_app().await;
+    let (token, _) = register_customer(&app, "fav-add-remove").await;
+    let admin = login_admin(&app).await;
+    let (_owner, shop_id, _product_id) = sellable(&app, &admin, "fav-add-remove", 1000).await;
+    assert_eq!(put_shop(&app, &token, &shop_id).await.status(), 200);
+
+    let mut lock_conn = app.db.acquire().await.unwrap();
+    sqlx::query("BEGIN").execute(&mut *lock_conn).await.unwrap();
+    sqlx::query("LOCK TABLE shop_profiles IN ACCESS EXCLUSIVE MODE")
+        .execute(&mut *lock_conn)
+        .await
+        .unwrap();
+
+    let add_base = app.base.clone();
+    let add_token = token.clone();
+    let add_shop_id = shop_id.clone();
+    let add = tokio::spawn(async move {
+        client()
+            .put(format!("{add_base}/api/favorites/shops/{add_shop_id}"))
+            .bearer_auth(add_token)
+            .send()
+            .await
+            .unwrap()
+    });
+
+    let hydration_waiting = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            let waiting: bool = sqlx::query_scalar(
+                "SELECT EXISTS(
+                   SELECT 1 FROM pg_stat_activity
+                    WHERE wait_event_type = 'Lock'
+                      AND query LIKE 'SELECT f.id, f.user_id, f.created_at,%'
+                 )",
+            )
+            .fetch_one(&app.db)
+            .await
+            .unwrap();
+            if waiting {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .is_ok();
+
+    let delete_base = app.base.clone();
+    let delete_token = token.clone();
+    let delete_shop_id = shop_id.clone();
+    let mut remove = tokio::spawn(async move {
+        client()
+            .delete(format!(
+                "{delete_base}/api/favorites/shops/{delete_shop_id}"
+            ))
+            .bearer_auth(delete_token)
+            .send()
+            .await
+            .unwrap()
+    });
+    let removed_before_hydration = tokio::time::timeout(Duration::from_millis(150), &mut remove)
+        .await
+        .is_ok();
+
+    sqlx::query("ROLLBACK")
+        .execute(&mut *lock_conn)
+        .await
+        .unwrap();
+    let add_response = add.await.unwrap();
+    let remove_response = if removed_before_hydration {
+        None
+    } else {
+        Some(remove.await.unwrap())
+    };
+
+    assert!(
+        hydration_waiting,
+        "favorite hydration never reached the table lock"
+    );
+    assert!(
+        !removed_before_hydration,
+        "favorite was removed before the idempotent add response was hydrated"
+    );
+    assert_eq!(
+        add_response.status(),
+        200,
+        "add: {:?}",
+        add_response.text().await
+    );
+    assert_eq!(remove_response.unwrap().status(), 204);
+}

apps/mall/pages/goods/[id].vue

@@ -251,7 +251,9 @@ const toggleFavorite = async (): Promise<void> => {
     }
   } catch (error) {
     if (error instanceof ApiError && error.status === 401) {
+      if (version === favoriteRequestVersion && product.value?.id === current.id) {
         await signInThenReturn();
+      }
       return;
     }
     if (version === favoriteRequestVersion && product.value?.id === current.id) {

apps/mall/pages/stores/[id].vue

@@ -138,7 +138,9 @@ const toggleFavorite = async (): Promise<void> => {
     }
   } catch (error) {
     if (error instanceof ApiError && error.status === 401) {
+      if (version === favoriteRequestVersion && store.value?.id === current.id) {
         await signInThenReturn();
+      }
       return;
     }
     if (version === favoriteRequestVersion && store.value?.id === current.id) {