Merge pull request #845 from golemfactory/wiezzel/custom_recv_addr

 Payment service: allow receiving funds without account init

core/model/src/payment.rs

@@ -125,6 +125,32 @@ pub mod local {
         }
     }
 
+    #[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)]
+    #[error("")]
+    pub struct NoError {} // This is needed because () doesn't implement Display
+
+    #[derive(Clone, Debug, Serialize, Deserialize)]
+    pub struct RegisterDriver {
+        pub driver_name: String,
+        pub platform: String,
+        pub recv_init_required: bool, // Is account initialization required for receiving payments
+    }
+
+    impl RpcMessage for RegisterDriver {
+        const ID: &'static str = "RegisterDriver";
+        type Item = ();
+        type Error = NoError;
+    }
+
+    #[derive(Clone, Debug, Serialize, Deserialize)]
+    pub struct UnregisterDriver(pub String);
+
+    impl RpcMessage for UnregisterDriver {
+        const ID: &'static str = "UnregisterDriver";
+        type Item = ();
+        type Error = NoError;
+    }
+
     #[derive(Clone, Debug, Serialize, Deserialize)]
     pub struct RegisterAccount {
         pub platform: String,
@@ -135,8 +161,10 @@ pub mod local {
 
     #[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)]
     pub enum RegisterAccountError {
-        #[error("Account already registered")]
-        AlreadyRegistered,
+        #[error("Account already registered: address={0}, driver={1}")]
+        AlreadyRegistered(String, String),
+        #[error("Driver not registered: {0}")]
+        DriverNotRegistered(String),
         #[error("Error while registering account: {0}")]
         Other(String),
     }
@@ -153,18 +181,10 @@ pub mod local {
         pub address: String,
     }
 
-    #[derive(Clone, Debug, Serialize, Deserialize, thiserror::Error)]
-    pub enum UnregisterAccountError {
-        #[error("Account not registered")]
-        NotRegistered,
-        #[error("Error while unregistering account: {0}")]
-        Other(String),
-    }
-
     impl RpcMessage for UnregisterAccount {
         const ID: &'static str = "UnregisterAccount";
         type Item = ();
-        type Error = UnregisterAccountError;
+        type Error = NoError;
     }
 
     #[derive(Clone, Debug, Serialize, Deserialize)]

core/payment-driver/base/src/bus.rs

@@ -1,10 +1,10 @@
 /*
-    Collection of interactions a PaymendDriver can have with ya_service_bus
+    Collection of interactions a PaymentDriver can have with ya_service_bus
 
     All interactions with the bus from the driver should go through this mod.
 */
 
-// Extrernal crates
+// External crates
 use std::sync::Arc;
 
 // Workspace uses
@@ -23,8 +23,11 @@ use ya_service_bus::{
 use crate::dao::DbExecutor;
 use crate::driver::PaymentDriver;
 
-pub async fn bind_service<Driver: PaymentDriver + 'static>(db: &DbExecutor, driver: Arc<Driver>) {
-    log::debug!("Binding payment driver service to service bus");
+pub async fn bind_service<Driver: PaymentDriver + 'static>(
+    db: &DbExecutor,
+    driver: Arc<Driver>,
+) -> anyhow::Result<()> {
+    log::debug!("Binding payment driver service to service bus...");
     let bus_id = driver_bus_id(driver.get_name());
 
     /* Short variable names explained:
@@ -33,8 +36,8 @@ pub async fn bind_service<Driver: PaymentDriver + 'static>(db: &DbExecutor, driv
         c = caller
         m = message
     */
-    #[rustfmt::skip] // Keep move's neatly alligned
-    ServiceBinder::new(&bus_id, db, driver)
+    #[rustfmt::skip] // Keep move's neatly aligned
+    ServiceBinder::new(&bus_id, db, driver.clone())
         .bind_with_processor(
             move |db, dr, c, m| async move { dr.init(db, c, m).await }
         )
@@ -57,14 +60,23 @@ pub async fn bind_service<Driver: PaymentDriver + 'static>(db: &DbExecutor, driv
             move |db, dr, c, m| async move { dr.validate_allocation(db, c, m).await }
         );
 
-    log::debug!("Successfully bound payment driver service to service bus");
-    log::debug!("Subscribing to identity events");
+    log::debug!("Successfully bound payment driver service to service bus.");
+
+    log::debug!("Subscribing to identity events...");
     let message = identity::Subscribe { endpoint: bus_id };
-    let result = service(identity::BUS_ID).send(message).await;
-    match result {
-        Err(e) => log::error!("init app-key listener error: {}", e),
-        _ => log::debug!("Successfully subscribed payment driver service to identity events"),
-    }
+    service(identity::BUS_ID).send(message).await??;
+    log::debug!("Successfully subscribed payment driver service to identity events.");
+
+    log::debug!("Registering driver in payment service...");
+    let message = payment_srv::RegisterDriver {
+        driver_name: driver.get_name(),
+        platform: driver.get_platform(),
+        recv_init_required: driver.recv_init_required(),
+    };
+    service(payment_srv::BUS_ID).send(message).await?.unwrap(); // Unwrap on purpose because it's NoError
+    log::debug!("Successfully registered driver in payment service.");
+
+    Ok(())
 }
 
 pub async fn list_unlocked_identities() -> Result<Vec<NodeId>, GenericError> {

core/payment-driver/base/src/driver.rs

@@ -35,6 +35,7 @@ pub trait PaymentDriver {
     // used by bus to bind service
     fn get_name(&self) -> String;
     fn get_platform(&self) -> String;
+    fn recv_init_required(&self) -> bool;
 
     async fn get_transaction_balance(
         &self,

core/payment-driver/dummy/src/lib.rs

@@ -8,6 +8,7 @@ pub struct PaymentDriverService;
 impl PaymentDriverService {
     pub async fn gsb<Context>(_context: &Context) -> anyhow::Result<()> {
         self::service::bind_service();
+        self::service::register_in_payment_service().await?;
         Ok(())
     }
 }

core/payment-driver/dummy/src/service.rs

@@ -6,6 +6,7 @@ use std::str::FromStr;
 use uuid::Uuid;
 use ya_core_model::driver::*;
 use ya_core_model::payment::local as payment_srv;
+use ya_service_bus::typed::service;
 use ya_service_bus::{typed as bus, RpcEndpoint};
 
 pub fn bind_service() {
@@ -22,6 +23,19 @@ pub fn bind_service() {
     log::debug!("Successfully bound payment driver service to service bus");
 }
 
+pub async fn register_in_payment_service() -> anyhow::Result<()> {
+    log::debug!("Registering driver in payment service...");
+    let message = payment_srv::RegisterDriver {
+        driver_name: DRIVER_NAME.to_string(),
+        platform: PLATFORM_NAME.to_string(),
+        recv_init_required: false,
+    };
+    service(payment_srv::BUS_ID).send(message).await?.unwrap(); // Unwrap on purpose because it's NoError
+    log::debug!("Successfully registered driver in payment service.");
+
+    Ok(())
+}
+
 async fn init(_db: (), _caller: String, msg: Init) -> Result<Ack, GenericError> {
     log::info!("init: {:?}", msg);
 

core/payment-driver/gnt/src/lib.rs

@@ -99,7 +99,8 @@ impl PaymentDriverService {
         let driver = GntDriver::new(db.clone()).await?;
         let processor = GNTDriverProcessor::new(driver);
         self::service::bind_service(&db, processor);
-        self::service::subscribe_to_identity_events().await;
+        self::service::subscribe_to_identity_events().await?;
+        self::service::register_in_payment_service().await?;
         Ok(())
     }
 }

core/payment-driver/gnt/src/service.rs

@@ -1,8 +1,10 @@
 use crate::processor::GNTDriverProcessor;
-use crate::DRIVER_NAME;
+use crate::{DRIVER_NAME, PLATFORM_NAME};
 use bigdecimal::BigDecimal;
 use ya_core_model::driver::*;
+use ya_core_model::payment::local as payment_srv;
 use ya_persistence::executor::DbExecutor;
+use ya_service_bus::typed::service;
 use ya_service_bus::{typed as bus, RpcEndpoint};
 
 pub fn bind_service(db: &DbExecutor, processor: GNTDriverProcessor) {
@@ -20,15 +22,26 @@ pub fn bind_service(db: &DbExecutor, processor: GNTDriverProcessor) {
     log::debug!("Successfully bound payment driver service to service bus");
 }
 
-pub async fn subscribe_to_identity_events() {
-    if let Err(e) = bus::service(ya_core_model::identity::BUS_ID)
+pub async fn subscribe_to_identity_events() -> anyhow::Result<()> {
+    bus::service(ya_core_model::identity::BUS_ID)
         .send(ya_core_model::identity::Subscribe {
             endpoint: driver_bus_id(DRIVER_NAME),
         })
-        .await
-    {
-        log::error!("init app-key listener error: {}", e)
-    }
+        .await??;
+    Ok(())
+}
+
+pub async fn register_in_payment_service() -> anyhow::Result<()> {
+    log::debug!("Registering driver in payment service...");
+    let message = payment_srv::RegisterDriver {
+        driver_name: DRIVER_NAME.to_string(),
+        platform: PLATFORM_NAME.to_string(),
+        recv_init_required: false,
+    };
+    service(payment_srv::BUS_ID).send(message).await?.unwrap(); // Unwrap on purpose because it's NoError
+    log::debug!("Successfully registered driver in payment service.");
+
+    Ok(())
 }
 
 async fn init(

core/payment-driver/zksync/src/driver.rs

@@ -130,6 +130,10 @@ impl PaymentDriver for ZksyncDriver {
         PLATFORM_NAME.to_string()
     }
 
+    fn recv_init_required(&self) -> bool {
+        false
+    }
+
     async fn get_transaction_balance(
         &self,
         _db: DbExecutor,

core/payment-driver/zksync/src/service.rs

@@ -35,7 +35,7 @@ impl ZksyncService {
         let driver = ZksyncDriver::new(db.clone());
         driver.load_active_accounts().await;
         let driver_rc = Arc::new(driver);
-        bus::bind_service(&db, driver_rc.clone()).await;
+        bus::bind_service(&db, driver_rc.clone()).await?;
         log::debug!("Driver loaded");
 
         // Start cron

core/payment/examples/payment_api.rs

@@ -195,6 +195,10 @@ async fn main() -> anyhow::Result<()> {
     ya_sb_router::bind_gsb_router(None).await?;
     log::debug!("bind_gsb_router()");
 
+    let processor = PaymentProcessor::new(db.clone());
+    ya_payment::service::bind_service(&db, processor);
+    log::debug!("bind_service()");
+
     let (driver_name, platform) = match args.driver {
         Driver::Dummy => {
             start_dummy_driver().await?;
@@ -210,10 +214,6 @@ async fn main() -> anyhow::Result<()> {
         }
     };
 
-    let processor = PaymentProcessor::new(db.clone());
-    ya_payment::service::bind_service(&db, processor);
-    log::debug!("bind_service()");
-
     bus::service(driver_bus_id(driver_name))
         .call(Init::new(provider_id.clone(), AccountMode::RECV))
         .await??;