Add remove opportunity broadcast for expired opportunities (#210)

auction-server/.clippy.toml

@@ -0,0 +1 @@
+ignore-interior-mutability = ["bytes::Bytes", "auction_server::opportunity::entities::opportunity::OpportunityKey", "ethers::types::Bytes"]

auction-server/src/api.rs

@@ -350,7 +350,9 @@ pub async fn start_api(run_options: RunOptions, store: Arc<StoreNew>) -> Result<
     opportunity::QuoteV1Svm,
     opportunity::OpportunityDelete,
     opportunity::OpportunityDeleteSvm,
+    opportunity::OpportunityDeleteEvm,
     opportunity::OpportunityDeleteV1Svm,
+    opportunity::OpportunityDeleteV1Evm,
     opportunity::ProgramSvm,
 
     ErrorBodyResponse,

auction-server/src/auction.rs

@@ -13,12 +13,15 @@ use {
             PermissionKeySvm,
         },
         models,
-        opportunity::service::{
-            get_live_opportunities::GetOpportunitiesInput,
-            ChainType as OpportunityChainType,
-            ChainTypeEvm as OpportunityChainTypeEvm,
-            ChainTypeSvm as OpportunityChainTypeSvm,
-            Service as OpportunityService,
+        opportunity::{
+            self,
+            service::{
+                get_live_opportunities::GetOpportunitiesInput,
+                ChainType as OpportunityChainType,
+                ChainTypeEvm as OpportunityChainTypeEvm,
+                ChainTypeSvm as OpportunityChainTypeSvm,
+                Service as OpportunityService,
+            },
         },
         server::{
             EXIT_CHECK_INTERVAL,
@@ -1224,7 +1227,10 @@ async fn verify_signatures_svm(
             let opportunities = store_new
                 .opportunity_service_svm
                 .get_live_opportunities(GetOpportunitiesInput {
-                    key: (bid.chain_id.clone(), permission_key.clone()),
+                    key: opportunity::entities::OpportunityKey(
+                        bid.chain_id.clone(),
+                        permission_key.clone(),
+                    ),
                 })
                 .await;
             opportunities.into_iter().any(|opportunity| {
@@ -1469,7 +1475,10 @@ pub trait ChainStore:
         !self
             .get_opportunity_service(store_new)
             .get_live_opportunities(GetOpportunitiesInput {
-                key: (self.get_name().clone(), permission_key.clone()),
+                key: opportunity::entities::OpportunityKey(
+                    self.get_name().clone(),
+                    permission_key.clone(),
+                ),
             })
             .await
             .is_empty()

auction-server/src/opportunity/api.rs

@@ -114,6 +114,18 @@ pub struct OpportunityDeleteV1Svm {
     pub program:            ProgramSvm,
 }
 
+/// Opportunity parameters needed for deleting live opportunities.
+#[serde_as]
+#[derive(Serialize, Deserialize, ToSchema, Clone, PartialEq, Debug)]
+pub struct OpportunityDeleteV1Evm {
+    /// The permission key of the opportunity.
+    #[schema(example = "0xdeadbeefcafe", value_type = String)]
+    pub permission_key: Bytes,
+    /// The chain id for the opportunity.
+    #[schema(example = "solana", value_type = String)]
+    pub chain_id:       ChainId,
+}
+
 #[derive(Serialize, Deserialize, ToSchema, Clone, PartialEq, Debug)]
 #[serde(tag = "version")]
 pub enum OpportunityDeleteSvm {
@@ -122,12 +134,24 @@ pub enum OpportunityDeleteSvm {
     V1(OpportunityDeleteV1Svm),
 }
 
+#[derive(Serialize, Deserialize, ToSchema, Clone, PartialEq, Debug)]
+#[serde(tag = "version")]
+pub enum OpportunityDeleteEvm {
+    #[serde(rename = "v1")]
+    #[schema(title = "v1")]
+    V1(OpportunityDeleteV1Evm),
+}
+
 /// The input type for deleting opportunities.
 #[derive(Serialize, Deserialize, ToSchema, Clone, PartialEq, Debug)]
-#[serde(untagged)]
+#[serde(tag = "chain_type")]
 pub enum OpportunityDelete {
+    #[serde(rename = "svm")]
     #[schema(title = "svm")]
     Svm(OpportunityDeleteSvm),
+    #[serde(rename = "evm")]
+    #[schema(title = "evm")]
+    Evm(OpportunityDeleteEvm),
 }
 
 /// The input type for creating a new opportunity.
@@ -608,6 +632,7 @@ impl OpportunityDelete {
     pub fn get_chain_id(&self) -> &ChainId {
         match self {
             OpportunityDelete::Svm(OpportunityDeleteSvm::V1(params)) => &params.chain_id,
+            OpportunityDelete::Evm(OpportunityDeleteEvm::V1(params)) => &params.chain_id,
         }
     }
 }
@@ -786,21 +811,26 @@ pub async fn delete_opportunities(
     State(store): State<Arc<StoreNew>>,
     Json(opportunity_delete): Json<OpportunityDelete>,
 ) -> Result<StatusCode, RestError> {
-    let OpportunityDelete::Svm(OpportunityDeleteSvm::V1(params)) = opportunity_delete;
-    if get_program(&auth)? != params.program {
-        return Err(RestError::Forbidden);
-    }
+    match opportunity_delete {
+        OpportunityDelete::Evm(_) => Err(RestError::Forbidden),
+        OpportunityDelete::Svm(params_svm) => {
+            let OpportunityDeleteSvm::V1(params) = params_svm;
+            if get_program(&auth)? != params.program {
+                return Err(RestError::Forbidden);
+            }
 
-    store
-        .opportunity_service_svm
-        .remove_opportunities(RemoveOpportunitiesInput {
-            chain_id:           params.chain_id,
-            permission_account: params.permission_account,
-            router:             params.router,
-        })
-        .await?;
+            store
+                .opportunity_service_svm
+                .remove_opportunities(RemoveOpportunitiesInput {
+                    chain_id:           params.chain_id,
+                    permission_account: params.permission_account,
+                    router:             params.router,
+                })
+                .await?;
 
-    Ok(StatusCode::NO_CONTENT)
+            Ok(StatusCode::NO_CONTENT)
+        }
+    }
 }
 
 pub fn get_routes(store: Arc<StoreNew>) -> Router<Arc<StoreNew>> {

auction-server/src/opportunity/entities/opportunity.rs

@@ -21,7 +21,9 @@ use {
 };
 
 pub type OpportunityId = Uuid;
-pub type OpportunityKey = (ChainId, PermissionKey);
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct OpportunityKey(pub ChainId, pub PermissionKey);
 
 #[derive(Debug, Clone, PartialEq)]
 pub struct OpportunityCoreFields<T: TokenAmount> {
@@ -73,8 +75,9 @@ pub trait Opportunity:
 
     fn new_with_current_time(val: Self::OpportunityCreate) -> Self;
     fn get_models_metadata(&self) -> Self::ModelMetadata;
+    fn get_opportunity_delete(&self) -> api::OpportunityDelete;
     fn get_key(&self) -> OpportunityKey {
-        (self.chain_id.clone(), self.permission_key.clone())
+        OpportunityKey(self.chain_id.clone(), self.permission_key.clone())
     }
 }
 

auction-server/src/opportunity/entities/opportunity_evm.rs

@@ -64,13 +64,20 @@ impl Opportunity for OpportunityEvm {
             target_calldata:   self.target_calldata.clone(),
         }
     }
+
+    fn get_opportunity_delete(&self) -> api::OpportunityDelete {
+        api::OpportunityDelete::Evm(api::OpportunityDeleteEvm::V1(api::OpportunityDeleteV1Evm {
+            permission_key: self.core_fields.permission_key.clone(),
+            chain_id:       self.core_fields.chain_id.clone(),
+        }))
+    }
 }
 
 impl OpportunityCreate for OpportunityCreateEvm {
     type ApiOpportunityCreate = api::OpportunityCreateEvm;
 
     fn get_key(&self) -> super::OpportunityKey {
-        (
+        super::OpportunityKey(
             self.core_fields.chain_id.clone(),
             self.core_fields.permission_key.clone(),
         )

auction-server/src/opportunity/entities/opportunity_svm.rs

@@ -105,13 +105,22 @@ impl Opportunity for OpportunitySvm {
             slot: self.slot,
         }
     }
+
+    fn get_opportunity_delete(&self) -> api::OpportunityDelete {
+        api::OpportunityDelete::Svm(api::OpportunityDeleteSvm::V1(api::OpportunityDeleteV1Svm {
+            chain_id:           self.chain_id.clone(),
+            permission_account: self.permission_account,
+            router:             self.router,
+            program:            self.program.clone().into(),
+        }))
+    }
 }
 
 impl OpportunityCreate for OpportunityCreateSvm {
     type ApiOpportunityCreate = api::OpportunityCreateSvm;
 
     fn get_key(&self) -> super::OpportunityKey {
-        (
+        super::OpportunityKey(
             self.core_fields.chain_id.clone(),
             self.core_fields.permission_key.clone(),
         )
@@ -299,15 +308,6 @@ impl OpportunitySvm {
     pub fn get_permission_key(router: Pubkey, permission_account: Pubkey) -> PermissionKey {
         PermissionKey::from([router.to_bytes(), permission_account.to_bytes()].concat())
     }
-
-    pub fn get_opportunity_delete(&self) -> api::OpportunityDelete {
-        api::OpportunityDelete::Svm(api::OpportunityDeleteSvm::V1(api::OpportunityDeleteV1Svm {
-            chain_id:           self.chain_id.clone(),
-            permission_account: self.permission_account,
-            router:             self.router,
-            program:            self.program.clone().into(),
-        }))
-    }
 }
 
 impl From<OpportunitySvmProgram> for api::ProgramSvm {

auction-server/src/opportunity/mod.rs

@@ -1,8 +1,8 @@
 mod contracts;
-mod entities;
 mod repository;
 mod token_spoof;
 
 pub mod api;
+pub mod entities;
 pub mod service;
 pub mod workers;

auction-server/src/opportunity/repository/remove_opportunity.rs

@@ -4,7 +4,10 @@ use {
         InMemoryStore,
         Repository,
     },
-    crate::opportunity::entities::Opportunity,
+    crate::opportunity::entities::{
+        self,
+        Opportunity,
+    },
     sqlx::Postgres,
     time::{
         OffsetDateTime,
@@ -17,8 +20,9 @@ impl<T: InMemoryStore> Repository<T> {
         &self,
         db: &sqlx::Pool<Postgres>,
         opportunity: &T::Opportunity,
-        reason: OpportunityRemovalReason,
+        reason: entities::OpportunityRemovalReason,
     ) -> anyhow::Result<()> {
+        let reason: OpportunityRemovalReason = reason.into();
         let now = OffsetDateTime::now_utc();
         sqlx::query("UPDATE opportunity SET removal_time = $1, removal_reason = $2 WHERE id = $3 AND removal_time IS NULL")
             .bind(PrimitiveDateTime::new(now.date(), now.time()))

auction-server/src/opportunity/service/get_quote.rs

@@ -218,7 +218,7 @@ impl Service<ChainTypeSvm> {
                         "Auction finished for the opportunity".to_string(),
                     ));
                 if let Err(e) = repo
-                    .remove_opportunity(&db, &opportunity, removal_reason.into())
+                    .remove_opportunity(&db, &opportunity, removal_reason)
                     .await
                 {
                     tracing::error!("Failed to remove opportunity: {:?}", e);

auction-server/src/opportunity/service/mod.rs

@@ -169,11 +169,17 @@ impl ConfigSvm {
     }
 }
 
+#[derive(Debug, PartialEq)]
+pub enum ChainTypeEnum {
+    Evm,
+    Svm,
+}
+
 pub trait ChainType: Send + Sync {
     type Config: Config;
     type InMemoryStore: InMemoryStore;
 
-    fn get_name() -> String;
+    fn get_type() -> ChainTypeEnum;
 }
 
 pub struct ChainTypeEvm;
@@ -183,17 +189,17 @@ impl ChainType for ChainTypeEvm {
     type Config = ConfigEvm;
     type InMemoryStore = InMemoryStoreEvm;
 
-    fn get_name() -> String {
-        "evm".to_string()
+    fn get_type() -> ChainTypeEnum {
+        ChainTypeEnum::Evm
     }
 }
 
 impl ChainType for ChainTypeSvm {
     type Config = ConfigSvm;
     type InMemoryStore = InMemoryStoreSvm;
 
-    fn get_name() -> String {
-        "svm".to_string()
+    fn get_type() -> ChainTypeEnum {
+        ChainTypeEnum::Svm
     }
 }
 

auction-server/src/opportunity/service/remove_invalid_or_expired_opportunities.rs

@@ -8,8 +8,17 @@ use {
         Service,
     },
     crate::{
-        api::RestError,
-        opportunity::entities,
+        api::{
+            ws::UpdateEvent,
+            RestError,
+        },
+        opportunity::{
+            entities::{
+                self,
+                Opportunity as _,
+            },
+            service::ChainTypeEnum,
+        },
         state::UnixTimestampMicros,
     },
     std::time::{
@@ -59,16 +68,46 @@ where
 
                 if let Some(reason) = reason {
                     tracing::info!(
-                        "Removing Opportunity {} for reason {:?}",
-                        opportunity.id,
-                        reason
+                        opportunity = ?opportunity,
+                        reason = ?reason,
+                        "Removing Opportunity",
                     );
-                    if let Err(e) = self
+                    match self
                         .repo
-                        .remove_opportunity(&self.db, opportunity, reason.into())
+                        .remove_opportunity(&self.db, opportunity, reason)
                         .await
                     {
-                        tracing::error!("Failed to remove opportunity: {}", e);
+                        Ok(()) => {
+                            // TODO Remove this later
+                            // For now we don't want searchers to update any of their code on EVM chains.
+                            // So we are not broadcasting remove opportunities event for EVM chains.
+                            if T::get_type() == ChainTypeEnum::Evm {
+                                continue;
+                            }
+
+                            // If there are no more opportunities with this key, it means all of the
+                            // opportunities have been removed for this key, so we can broadcast remove opportunities event.
+                            if self
+                                .repo
+                                .get_in_memory_opportunities_by_key(&opportunity.get_key())
+                                .await
+                                .is_empty()
+                            {
+                                if let Err(e) = self.store.ws.broadcast_sender.send(
+                                    UpdateEvent::RemoveOpportunities(
+                                        opportunity.get_opportunity_delete(),
+                                    ),
+                                ) {
+                                    tracing::error!(
+                                        error = e.to_string(),
+                                        "Failed to broadcast remove opportunity"
+                                    );
+                                }
+                            }
+                        }
+                        Err(e) => {
+                            tracing::error!(error = ?e, "Failed to remove opportunity");
+                        }
                     }
                 }
             }