active listing constraint for bidding

migrations/02_add_bid_active_listing_constraint.sql

@@ -0,0 +1,15 @@
+-- Add constraint to prevent bids on inactive listings
+-- This ensures atomicity at the database level
+
+-- Create a trigger to prevent bids on inactive listings
+-- This is more reliable than CHECK constraints in SQLite
+CREATE TRIGGER prevent_bid_on_inactive_listing
+    BEFORE INSERT ON bids
+    FOR EACH ROW
+    WHEN (
+        SELECT COALESCE(is_active, 0) FROM listings WHERE id = NEW.listing_id
+    ) != 1
+BEGIN
+    SELECT RAISE(ABORT, 'Cannot place bid on inactive listing');
+END;
+

src/db/dao/bid_dao.rs

@@ -20,6 +20,20 @@ impl BidDAO {
 #[allow(unused)]
 impl BidDAO {
     pub async fn insert_bid(&self, bid: &NewBid) -> Result<PersistedBid> {
+        let mut tx = self.0.begin().await?;
+
+        // First, validate that the listing is active within the transaction
+        // Note: SQLite doesn't support FOR UPDATE, but transactions provide isolation
+        let is_active: bool = sqlx::query_scalar("SELECT is_active FROM listings WHERE id = ?")
+            .bind(bid.listing_id)
+            .fetch_optional(&mut *tx)
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("Listing not found"))?;
+
+        if !is_active {
+            return Err(anyhow::anyhow!("Cannot place bid on inactive listing"));
+        }
+
         let now = Utc::now();
         let binds = BindFields::default()
             .push("listing_id", &bid.listing_id)
@@ -32,7 +46,7 @@ impl BidDAO {
             .push("created_at", &now)
             .push("updated_at", &now);
 
-        let query_str = format!(
+        let insert_bid_query_str = format!(
             r#"
                 INSERT INTO bids ({}) VALUES ({})
                 RETURNING *
@@ -40,12 +54,11 @@ impl BidDAO {
             binds.bind_names().join(", "),
             binds.bind_placeholders().join(", ")
         );
+        let insert_bid_query = binds.bind_to_query(sqlx::query(&insert_bid_query_str));
 
-        let row = binds
-            .bind_to_query(sqlx::query(&query_str))
-            .fetch_one(&self.0)
-            .await?;
-        Ok(FromRow::from_row(&row)?)
+        let row = insert_bid_query.fetch_one(&mut *tx).await?;
+        tx.commit().await?;
+        Ok(PersistedBid::from_row(&row)?)
     }
 
     pub async fn bidder_ids_for_listing(&self, listing_id: DbListingId) -> Result<Vec<DbUserId>> {
@@ -171,7 +184,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_insert_bid() {
+    async fn test_insert_bid_on_active_listing() {
         let (_user_dao, _listing_dao, bid_dao, user_id, listing_id) =
             create_test_user_and_listing().await;
 
@@ -186,11 +199,11 @@ mod tests {
             proxy_bid_id: None,
         };
 
-        // Insert bid
+        // Insert bid on active listing should succeed
         let inserted_bid = bid_dao
             .insert_bid(&new_bid)
             .await
-            .expect("Failed to insert bid");
+            .expect("Failed to insert bid on active listing");
 
         // Verify the inserted bid has the correct values
         assert_eq!(inserted_bid.listing_id, new_bid.listing_id);
@@ -210,4 +223,50 @@ mod tests {
             inserted_bid.persisted.updated_at
         );
     }
+
+    #[tokio::test]
+    async fn test_insert_bid_on_inactive_listing_fails() {
+        let (_user_dao, listing_dao, bid_dao, user_id, listing_id) =
+            create_test_user_and_listing().await;
+
+        // Mark the listing as inactive
+
+        listing_dao
+            .set_listing_is_active(listing_id, false)
+            .await
+            .unwrap();
+
+        let new_bid = NewBid {
+            persisted: (),
+            listing_id,
+            buyer_id: user_id,
+            bid_amount: MoneyAmount::from_str("25.50").unwrap(),
+            description: Some("Test bid description".to_string()),
+            is_cancelled: false,
+            slot_number: Some(1),
+            proxy_bid_id: None,
+        };
+
+        // Verify the listing is actually inactive
+        let updated_listing = listing_dao.find_by_id(listing_id).await.unwrap().unwrap();
+        assert!(
+            !updated_listing.base.is_active,
+            "Listing should be inactive"
+        );
+
+        // Insert bid on inactive listing should fail
+        let result = bid_dao.insert_bid(&new_bid).await;
+        match result {
+            Ok(_) => panic!("Expected bid insertion to fail on inactive listing, but it succeeded"),
+            Err(e) => {
+                let error_msg = e.to_string();
+                assert!(
+                    error_msg.contains("inactive listing")
+                        || error_msg.contains("Cannot place bid"),
+                    "Expected error about inactive listing, got: {}",
+                    error_msg
+                );
+            }
+        }
+    }
 }

src/db/dao/listing_dao.rs

@@ -172,7 +172,7 @@ impl ListingDAO {
         let result = sqlx::query_as(
             r#"
             SELECT * FROM listings
-            AND is_active = 1
+            WHERE is_active = 1
             ORDER BY ends_at ASC
             LIMIT 1"#,
         )
@@ -195,6 +195,7 @@ fn binds_for_base(base: &ListingBase) -> BindFields {
         .push("currency_type", &base.currency_type)
         .push("starts_at", &base.starts_at)
         .push("ends_at", &base.ends_at)
+        .push("is_active", &base.is_active)
 }
 
 fn binds_for_fields(fields: &ListingFields) -> BindFields {
@@ -272,3 +273,175 @@ impl FromRow<'_, SqliteRow> for PersistedListing {
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::test_utils::{create_deps, with_test_listing, with_test_user};
+    use chrono::{Duration, Utc};
+    use rstest::rstest;
+
+    #[tokio::test]
+    async fn test_find_next_ending_listing_no_active_listings() {
+        let deps = create_deps().await;
+        let listing_dao = &deps.deps.get::<crate::db::DAOs>().listing;
+
+        let result = listing_dao
+            .find_next_ending_listing()
+            .await
+            .expect("Failed to query for next ending listing");
+
+        assert!(
+            result.is_none(),
+            "Expected no listings when database is empty"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_find_next_ending_listing_single_active_listing() {
+        let deps = create_deps().await;
+        let listing_dao = &deps.deps.get::<crate::db::DAOs>().listing;
+
+        let seller = with_test_user(&deps.deps, |_| {}).await;
+        let ends_at = Utc::now() + Duration::hours(1);
+
+        let inserted_listing = with_test_listing(&deps.deps, &seller, |listing| {
+            listing.base.ends_at = ends_at;
+        })
+        .await;
+
+        let result = listing_dao
+            .find_next_ending_listing()
+            .await
+            .expect("Failed to query for next ending listing");
+
+        assert!(result.is_some(), "Expected to find the active listing");
+        let found_listing = result.unwrap();
+        assert_eq!(found_listing.persisted.id, inserted_listing.persisted.id);
+        assert_eq!(found_listing.base.ends_at, ends_at);
+        assert!(found_listing.base.is_active);
+    }
+
+    #[rstest]
+    #[case(3, vec![2, 1, 3], 1)] // Multiple listings, should return the one ending soonest (index 1)
+    #[case(2, vec![1, 2], 0)] // Two listings, should return the one ending first (index 0)
+    #[case(4, vec![4, 1, 3, 2], 1)] // Four listings, should return the one ending soonest (index 1)
+    #[tokio::test]
+    async fn test_find_next_ending_listing_multiple_active_listings(
+        #[case] _num_listings: usize,
+        #[case] hours_from_now: Vec<i64>,
+        #[case] expected_index: usize,
+    ) {
+        let deps = create_deps().await;
+        let listing_dao = &deps.deps.get::<crate::db::DAOs>().listing;
+        let seller = with_test_user(&deps.deps, |_| {}).await;
+
+        let mut inserted_listings = Vec::new();
+        let base_time = Utc::now();
+
+        // Create multiple listings with different end times
+        for (_i, hours) in hours_from_now.iter().enumerate() {
+            let ends_at = base_time + Duration::hours(*hours);
+            let inserted_listing = with_test_listing(&deps.deps, &seller, |listing| {
+                listing.base.ends_at = ends_at;
+            })
+            .await;
+            inserted_listings.push(inserted_listing);
+        }
+
+        let result = listing_dao
+            .find_next_ending_listing()
+            .await
+            .expect("Failed to query for next ending listing");
+
+        assert!(result.is_some(), "Expected to find an active listing");
+        let found_listing = result.unwrap();
+
+        // Should return the listing that ends soonest
+        let expected_listing = &inserted_listings[expected_index];
+        assert_eq!(found_listing.persisted.id, expected_listing.persisted.id);
+        assert_eq!(found_listing.base.ends_at, expected_listing.base.ends_at);
+    }
+
+    #[tokio::test]
+    async fn test_find_next_ending_listing_ignores_inactive_listings() {
+        let deps = create_deps().await;
+        let listing_dao = &deps.deps.get::<crate::db::DAOs>().listing;
+        let seller = with_test_user(&deps.deps, |_| {}).await;
+
+        // Create an inactive listing that ends sooner
+        let inactive_ends_at = Utc::now() + Duration::hours(1);
+        with_test_listing(&deps.deps, &seller, |listing| {
+            listing.base.ends_at = inactive_ends_at;
+            listing.base.is_active = false;
+        })
+        .await;
+
+        // Create an active listing that ends later
+        let active_ends_at = Utc::now() + Duration::hours(2);
+        let inserted_active_listing = with_test_listing(&deps.deps, &seller, |listing| {
+            listing.base.ends_at = active_ends_at;
+        })
+        .await;
+
+        let result = listing_dao
+            .find_next_ending_listing()
+            .await
+            .expect("Failed to query for next ending listing");
+
+        assert!(result.is_some(), "Expected to find the active listing");
+        let found_listing = result.unwrap();
+
+        // Should return the active listing, not the inactive one that ends sooner
+        assert_eq!(
+            found_listing.persisted.id,
+            inserted_active_listing.persisted.id
+        );
+        assert_eq!(found_listing.base.ends_at, active_ends_at);
+        assert!(found_listing.base.is_active);
+    }
+
+    #[tokio::test]
+    async fn test_find_next_ending_listing_with_mixed_active_inactive() {
+        let deps = create_deps().await;
+        let listing_dao = &deps.deps.get::<crate::db::DAOs>().listing;
+        let seller = with_test_user(&deps.deps, |_| {}).await;
+
+        let base_time = Utc::now();
+        let mut inserted_active_listings = Vec::new();
+
+        // Create a mix of active and inactive listings
+        let listing_configs = vec![
+            (1, false), // inactive, ends in 1 hour
+            (2, true),  // active, ends in 2 hours
+            (3, false), // inactive, ends in 3 hours
+            (4, true),  // active, ends in 4 hours
+            (5, true),  // active, ends in 5 hours
+        ];
+
+        for (hours, is_active) in listing_configs {
+            let ends_at = base_time + Duration::hours(hours);
+            let inserted_listing = with_test_listing(&deps.deps, &seller, |listing| {
+                listing.base.ends_at = ends_at;
+                listing.base.is_active = is_active;
+            })
+            .await;
+
+            if is_active {
+                inserted_active_listings.push(inserted_listing);
+            }
+        }
+
+        let result = listing_dao
+            .find_next_ending_listing()
+            .await
+            .expect("Failed to query for next ending listing");
+
+        assert!(result.is_some(), "Expected to find an active listing");
+        let found_listing = result.unwrap();
+
+        // Should return the first active listing (ends in 2 hours), ignoring inactive ones
+        let expected_listing = &inserted_active_listings[0]; // The one ending in 2 hours
+        assert_eq!(found_listing.persisted.id, expected_listing.persisted.id);
+        assert!(found_listing.base.is_active);
+    }
+}

src/listing_expiry_checker_task.rs

@@ -1,3 +1,5 @@
+use chrono::{DateTime, Duration, Utc};
+use std::collections::VecDeque;
 use tokio::sync::mpsc::Receiver;
 
 use crate::{
@@ -10,11 +12,41 @@ pub async fn task(
     daos: DAOs,
     mut listing_event: Receiver<ListingUpdatedEvent>,
 ) -> BotResult<()> {
+    const ERROR_WINDOW: Duration = Duration::seconds(3);
+    const COOLDOWN_DELAY: Duration = Duration::seconds(3);
+
+    let mut error_timestamps: VecDeque<DateTime<Utc>> = VecDeque::new();
+
     loop {
         match task_impl(&message_sender, &daos, &mut listing_event).await {
             Ok(r) => return Ok(r),
             Err(e) => {
                 log::error!("Error in listing expiry checker task: {e}");
+
+                let now = Utc::now();
+                error_timestamps.push_back(now);
+
+                // Remove errors older than the error window
+                while let Some(&front_time) = error_timestamps.front() {
+                    if now - front_time > ERROR_WINDOW {
+                        error_timestamps.pop_front();
+                    } else {
+                        break;
+                    }
+                }
+
+                // If we have 2 or more errors within the window, apply cooldown
+                if error_timestamps.len() >= 2 {
+                    log::warn!(
+                        "Multiple errors ({}) within {} seconds, applying cooldown of {} seconds",
+                        error_timestamps.len(),
+                        ERROR_WINDOW.num_seconds(),
+                        COOLDOWN_DELAY.num_seconds()
+                    );
+
+                    tokio::time::sleep(COOLDOWN_DELAY.to_std().map_err(BotError::internal)?).await;
+                }
+
                 continue;
             }
         }