Account controller zknym command request handling

nym-vpn-core/Cargo.toml

@@ -123,6 +123,7 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tun = { version = "0.6.1", features = ["async"] }
 uniffi = { version = "0.27.3", features = ["cli"] }
 url = "2.5"
+uuid = "1.11"
 vergen = { version = "8.3.1", default-features = false }
 windows-sys = "0.52"
 x25519-dalek = "2.0"
@@ -135,6 +136,7 @@ nym-client-core = { git = "https://github.com/nymtech/nym", branch = "release/20
 nym-compact-ecash = { git = "https://github.com/nymtech/nym", branch = "release/2024.13-magura" }
 nym-config = { git = "https://github.com/nymtech/nym", branch = "release/2024.13-magura" }
 nym-contracts-common = { git = "https://github.com/nymtech/nym", branch = "release/2024.13-magura" }
+nym-credential-proxy-requests = { git = "https://github.com/nymtech/nym", branch = "release/2024.13-magura" }
 nym-credential-storage = { git = "https://github.com/nymtech/nym", branch = "release/2024.13-magura" }
 nym-credentials = { git = "https://github.com/nymtech/nym", branch = "release/2024.13-magura" }
 nym-credentials-interface = { git = "https://github.com/nymtech/nym", branch = "release/2024.13-magura" }

nym-vpn-core/crates/nym-vpn-account-controller/Cargo.toml

@@ -9,8 +9,11 @@ edition.workspace = true
 license.workspace = true
 
 [dependencies]
+futures.workspace = true
+# nym-client-core = { workspace = true, features = ["cli", "fs-credentials-storage", "fs-surb-storage", "fs-gateways-storage"] }
 nym-compact-ecash.workspace = true
 nym-config.workspace = true
+nym-credential-proxy-requests.workspace = true
 nym-credential-storage.workspace = true
 nym-credentials-interface.workspace = true
 nym-credentials.workspace = true
@@ -31,3 +34,6 @@ tokio-util.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 url.workspace = true
+
+uuid = "1.11"
+si-scale = "0.2.3"

nym-vpn-core/crates/nym-vpn-account-controller/src/commands/mod.rs

@@ -0,0 +1,159 @@
+// Copyright 2024 - Nym Technologies SA <[email protected]>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use nym_vpn_api_client::{
+    response::NymVpnDevice,
+    types::{Device, VpnApiAccount},
+    VpnApiClient,
+};
+
+use crate::{
+    controller::{AccountSummaryResponse, DevicesResponse, PendingCommands},
+    error::Error,
+    SharedAccountState,
+};
+
+pub(crate) mod register_device;
+pub(crate) mod update_state;
+pub(crate) mod zknym;
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum AccountCommand {
+    UpdateAccountState,
+    RegisterDevice,
+    RequestZkNym,
+    GetDeviceZkNym,
+    GetZkNymsAvailableForDownload,
+    GetZkNymById(String),
+    GetAvailableTickets,
+}
+
+#[derive(Clone, Debug)]
+pub(crate) enum AccountCommandResult {
+    UpdatedAccountState,
+    RegisteredDevice(NymVpnDevice),
+}
+
+pub(crate) struct CommandHandler {
+    id: uuid::Uuid,
+    command: AccountCommand,
+
+    account: VpnApiAccount,
+    device: Device,
+    pending_command: PendingCommands,
+    account_state: SharedAccountState,
+    vpn_api_client: VpnApiClient,
+
+    last_account_summary: AccountSummaryResponse,
+    last_devices: DevicesResponse,
+}
+
+impl CommandHandler {
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn new(
+        command: AccountCommand,
+        account: VpnApiAccount,
+        device: Device,
+        pending_command: PendingCommands,
+        account_state: SharedAccountState,
+        vpn_api_client: VpnApiClient,
+        last_account_summary: AccountSummaryResponse,
+        last_devices: DevicesResponse,
+    ) -> Self {
+        let id = uuid::Uuid::new_v4();
+        pending_command
+            .lock()
+            .map(|mut guard| guard.insert(id, command.clone()))
+            .map_err(|err| {
+                tracing::error!(
+                    "Failed to insert command {} into pending commands: {:?}",
+                    id,
+                    err
+                )
+            })
+            .ok();
+        tracing::debug!("Created command handler with id: {}", id);
+        CommandHandler {
+            id,
+            command,
+            account,
+            device,
+            pending_command,
+            account_state,
+            vpn_api_client,
+            last_account_summary,
+            last_devices,
+        }
+    }
+
+    async fn update_shared_account_state(&self) -> Result<AccountCommandResult, Error> {
+        let update_result = update_state::update_state(
+            &self.account,
+            &self.device,
+            &self.account_state,
+            &self.vpn_api_client,
+            &self.last_account_summary,
+            &self.last_devices,
+        )
+        .await
+        .map(|_| AccountCommandResult::UpdatedAccountState);
+        tracing::debug!("Current state: {:?}", self.account_state.lock().await);
+        update_result
+    }
+
+    async fn register_device(&self) -> Result<AccountCommandResult, Error> {
+        register_device::register_device(
+            &self.account,
+            &self.device,
+            &self.account_state,
+            &self.vpn_api_client,
+        )
+        .await
+        .map(AccountCommandResult::RegisteredDevice)
+    }
+
+    pub(crate) async fn run(self) -> Result<AccountCommandResult, Error> {
+        tracing::debug!("Running command {:?} with id {}", self.command, self.id);
+        match self.command {
+            AccountCommand::UpdateAccountState => self.update_shared_account_state().await,
+            AccountCommand::RegisterDevice => self.register_device().await,
+            AccountCommand::RequestZkNym => todo!(),
+            AccountCommand::GetDeviceZkNym => todo!(),
+            AccountCommand::GetZkNymsAvailableForDownload => todo!(),
+            AccountCommand::GetZkNymById(_) => todo!(),
+            AccountCommand::GetAvailableTickets => todo!(),
+        }
+        .inspect(|_result| {
+            tracing::info!("Command {:?} with id {} completed", self.command, self.id);
+        })
+        .inspect_err(|err| {
+            tracing::warn!(
+                "Command {:?} with id {} completed with error",
+                self.command,
+                self.id
+            );
+            tracing::debug!(
+                "Command {:?} with id {} failed with error: {:?}",
+                self.command,
+                self.id,
+                err
+            );
+        })
+    }
+}
+
+impl Drop for CommandHandler {
+    fn drop(&mut self) {
+        self.pending_command
+            .lock()
+            .map(|mut guard| guard.remove(&self.id))
+            .inspect_err(|err| {
+                tracing::error!(
+                    "Failed to remove command {} from pending commands: {:?}",
+                    self.id,
+                    err
+                )
+            })
+            .ok();
+    }
+}

nym-vpn-core/crates/nym-vpn-account-controller/src/commands/register_device.rs

@@ -0,0 +1,29 @@
+// Copyright 2024 - Nym Technologies SA <[email protected]>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use nym_vpn_api_client::{
+    response::NymVpnDevice,
+    types::{Device, VpnApiAccount},
+};
+
+use crate::{error::Error, shared_state::DeviceState, SharedAccountState};
+
+pub(crate) async fn register_device(
+    account: &VpnApiAccount,
+    device: &Device,
+    account_state: &SharedAccountState,
+    vpn_api_client: &nym_vpn_api_client::VpnApiClient,
+) -> Result<NymVpnDevice, Error> {
+    let response = vpn_api_client
+        .register_device(account, device)
+        .await
+        .inspect(|device_result| {
+            tracing::info!("Response: {:#?}", device_result);
+            tracing::info!("Device registered: {}", device_result.device_identity_key);
+        })
+        .map_err(Error::RegisterDevice)?;
+
+    let device_state = DeviceState::from(response.status);
+    account_state.set_device(device_state).await;
+    Ok(response)
+}

nym-vpn-core/crates/nym-vpn-account-controller/src/commands/update_state.rs

@@ -0,0 +1,155 @@
+// Copyright 2024 - Nym Technologies SA <[email protected]>
+// SPDX-License-Identifier: GPL-3.0-only
+
+use std::sync::Arc;
+
+use nym_vpn_api_client::{
+    response::{NymErrorResponse, NymVpnAccountSummaryResponse, NymVpnDevicesResponse},
+    types::{Device, VpnApiAccount},
+    HttpClientError,
+};
+
+use crate::{
+    error::Error,
+    shared_state::{AccountState, DeviceState, SharedAccountState, SubscriptionState},
+};
+
+pub(crate) async fn update_state(
+    account: &VpnApiAccount,
+    device: &Device,
+    account_state: &SharedAccountState,
+    vpn_api_client: &nym_vpn_api_client::VpnApiClient,
+    last_account_summary: &Arc<tokio::sync::Mutex<Option<NymVpnAccountSummaryResponse>>>,
+    last_devices: &Arc<tokio::sync::Mutex<Option<NymVpnDevicesResponse>>>,
+) -> Result<(), Error> {
+    update_account_state(account, account_state, vpn_api_client, last_account_summary).await?;
+    update_device_state(account, device, account_state, vpn_api_client, last_devices).await?;
+    get_zk_nym_status(account, device, vpn_api_client).await?;
+    Ok(())
+}
+
+async fn update_account_state(
+    account: &VpnApiAccount,
+    account_state: &SharedAccountState,
+    vpn_api_client: &nym_vpn_api_client::VpnApiClient,
+    last_account_summary: &Arc<tokio::sync::Mutex<Option<NymVpnAccountSummaryResponse>>>,
+) -> Result<(), Error> {
+    tracing::debug!("Updating account state");
+    let response = vpn_api_client.get_account_summary(account).await;
+
+    // Check if the response indicates that we are not registered
+    if let Some(403) = &response.as_ref().err().and_then(extract_status_code) {
+        tracing::warn!("NymVPN API reports: access denied (403)");
+        account_state.set_account(AccountState::NotRegistered).await;
+    }
+
+    let account_summary = response.map_err(|source| {
+        tracing::warn!("NymVPN API error response: {:?}", source);
+        Error::GetAccountSummary {
+            base_url: vpn_api_client.current_url().clone(),
+            source: Box::new(source),
+        }
+    })?;
+
+    if last_account_summary
+        .lock()
+        .await
+        .replace(account_summary.clone())
+        .as_ref()
+        != Some(&account_summary)
+    {
+        tracing::info!("Account summary: {:#?}", account_summary);
+    }
+
+    account_state
+        .set_account(AccountState::from(account_summary.account.status))
+        .await;
+
+    account_state
+        .set_subscription(SubscriptionState::from(account_summary.subscription))
+        .await;
+
+    Ok(())
+}
+
+async fn update_device_state(
+    account: &VpnApiAccount,
+    our_device: &Device,
+    account_state: &SharedAccountState,
+    vpn_api_client: &nym_vpn_api_client::VpnApiClient,
+    last_devices: &Arc<tokio::sync::Mutex<Option<NymVpnDevicesResponse>>>,
+) -> Result<(), Error> {
+    tracing::debug!("Updating device state");
+
+    let devices = vpn_api_client
+        .get_devices(account)
+        .await
+        .map_err(Error::GetDevices)?;
+
+    if last_devices.lock().await.replace(devices.clone()).as_ref() != Some(&devices) {
+        tracing::info!("Registered devices: {}", devices);
+    }
+
+    // TODO: pagination
+    let found_device = devices
+        .items
+        .iter()
+        .find(|device| device.device_identity_key == our_device.identity_key().to_base58_string());
+
+    let Some(found_device) = found_device else {
+        tracing::info!("Our device is not registered");
+        account_state.set_device(DeviceState::NotRegistered).await;
+        return Ok(());
+    };
+
+    account_state
+        .set_device(DeviceState::from(found_device.status))
+        .await;
+
+    Ok(())
+}
+
+async fn get_zk_nym_status(
+    account: &VpnApiAccount,
+    device: &Device,
+    vpn_api_client: &nym_vpn_api_client::VpnApiClient,
+) -> Result<(), Error> {
+    tracing::debug!("Getting ZK nym status");
+
+    tracing::info!("Checking device zk nyms");
+    let response = vpn_api_client.get_device_zk_nyms(account, device).await;
+    tracing::info!("{:#?}", response);
+
+    tracing::info!("Checking zk nyms available for download");
+    let response = vpn_api_client
+        .get_zk_nyms_available_for_download(account, device)
+        .await;
+    tracing::info!("{:#?}", response);
+    Ok(())
+}
+
+fn extract_status_code<E>(err: &E) -> Option<u16>
+where
+    E: std::error::Error + 'static,
+{
+    let mut source = err.source();
+    while let Some(err) = source {
+        if let Some(status) = err
+            .downcast_ref::<HttpClientError<NymErrorResponse>>()
+            .and_then(extract_status_code_inner)
+        {
+            return Some(status);
+        }
+        source = err.source();
+    }
+    None
+}
+
+fn extract_status_code_inner(
+    err: &nym_vpn_api_client::HttpClientError<NymErrorResponse>,
+) -> Option<u16> {
+    match err {
+        HttpClientError::EndpointFailure { status, .. } => Some((*status).into()),
+        _ => None,
+    }
+}