fwat: subscribe to token updates

Cargo.toml

@@ -7,14 +7,14 @@ version = "1.0.9"
 crate-type = ["cdylib", "rlib", "staticlib"]
 
 [dependencies]
-dojo-world = { git = "https://github.com/dojoengine/dojo", rev = "d11a7ec" }
-dojo-types = { git = "https://github.com/dojoengine/dojo", rev = "d11a7ec"}
-torii-client = { git = "https://github.com/dojoengine/dojo", rev = "d11a7ec" }
+dojo-world = { git = "https://github.com/dojoengine/dojo", rev = "1aa06e1" }
+dojo-types = { git = "https://github.com/dojoengine/dojo", rev = "1aa06e1"}
+torii-client = { git = "https://github.com/dojoengine/dojo", rev = "1aa06e1" }
 torii-grpc = { git = "https://github.com/dojoengine/dojo", features = [
     "client",
-], rev = "d11a7ec" }
-torii-typed-data = { git = "https://github.com/dojoengine/dojo", rev = "d11a7ec" }
-torii-relay = { git = "https://github.com/dojoengine/dojo", rev = "d11a7ec" }
+], rev = "1aa06e1" }
+torii-typed-data = { git = "https://github.com/dojoengine/dojo", rev = "1aa06e1" }
+torii-relay = { git = "https://github.com/dojoengine/dojo", rev = "1aa06e1" }
 
 starknet = "0.12.0"
 starknet-crypto = "0.7.2"

dojo.h

@@ -266,6 +266,15 @@ typedef struct ResultCArrayToken {
   };
 } ResultCArrayToken;
 
+typedef struct Token {
+  const char *id;
+  struct FieldElement contract_address;
+  const char *name;
+  const char *symbol;
+  uint8_t decimals;
+  const char *metadata;
+} Token;
+
 typedef struct CArrayTokenBalance {
   struct TokenBalance *data;
   uintptr_t data_len;
@@ -731,14 +740,6 @@ typedef struct EntityKeysClause {
   };
 } EntityKeysClause;
 
-typedef struct Token {
-  struct FieldElement contract_address;
-  const char *name;
-  const char *symbol;
-  uint8_t decimals;
-  const char *metadata;
-} Token;
-
 typedef enum COptionFieldElement_Tag {
   SomeFieldElement,
   NoneFieldElement,
@@ -1122,6 +1123,22 @@ struct ResultCArrayToken client_tokens(struct ToriiClient *client,
                                        const struct FieldElement *contract_addresses,
                                        uintptr_t contract_addresses_len);
 
+/**
+ * Subscribes to token updates
+ *
+ * # Parameters
+ * * `client` - Pointer to ToriiClient instance
+ * * `contract_addresses` - Array of contract addresses
+ * * `callback` - Function called when updates occur
+ *
+ * # Returns
+ * Result containing pointer to Subscription or error
+ */
+struct ResultSubscription client_on_token_update(struct ToriiClient *client,
+                                                 const struct FieldElement *contract_addresses,
+                                                 uintptr_t contract_addresses_len,
+                                                 void (*callback)(struct Token));
+
 /**
  * Gets token balances for given accounts and contracts
  *

dojo.hpp

@@ -593,6 +593,7 @@ struct Event {
 };
 
 struct Token {
+  const char *id;
   FieldElement contract_address;
   const char *name;
   const char *symbol;
@@ -1237,6 +1238,20 @@ Result<CArray<Token>> client_tokens(ToriiClient *client,
                                     const FieldElement *contract_addresses,
                                     uintptr_t contract_addresses_len);
 
+/// Subscribes to token updates
+///
+/// # Parameters
+/// * `client` - Pointer to ToriiClient instance
+/// * `contract_addresses` - Array of contract addresses
+/// * `callback` - Function called when updates occur
+///
+/// # Returns
+/// Result containing pointer to Subscription or error
+Result<Subscription*> client_on_token_update(ToriiClient *client,
+                                             const FieldElement *contract_addresses,
+                                             uintptr_t contract_addresses_len,
+                                             void (*callback)(Token));
+
 /// Gets token balances for given accounts and contracts
 ///
 /// # Parameters

dojo.pyx

@@ -171,6 +171,14 @@ cdef extern from *:
     CArrayToken ok;
     Error err;
 
+  cdef struct Token:
+    const char *id;
+    FieldElement contract_address;
+    const char *name;
+    const char *symbol;
+    uint8_t decimals;
+    const char *metadata;
+
   cdef struct CArrayTokenBalance:
     TokenBalance *data;
     uintptr_t data_len;
@@ -453,13 +461,6 @@ cdef extern from *:
     CArrayFieldElement hashed_keys;
     KeysClause entity_keys;
 
-  cdef struct Token:
-    FieldElement contract_address;
-    const char *name;
-    const char *symbol;
-    uint8_t decimals;
-    const char *metadata;
-
   cdef enum COptionFieldElement_Tag:
     SomeFieldElement,
     NoneFieldElement,
@@ -784,6 +785,20 @@ cdef extern from *:
                                   const FieldElement *contract_addresses,
                                   uintptr_t contract_addresses_len);
 
+  # Subscribes to token updates
+  #
+  # # Parameters
+  # * `client` - Pointer to ToriiClient instance
+  # * `contract_addresses` - Array of contract addresses
+  # * `callback` - Function called when updates occur
+  #
+  # # Returns
+  # Result containing pointer to Subscription or error
+  ResultSubscription client_on_token_update(ToriiClient *client,
+                                            const FieldElement *contract_addresses,
+                                            uintptr_t contract_addresses_len,
+                                            void (*callback)(Token));
+
   # Gets token balances for given accounts and contracts
   #
   # # Parameters

src/c/mod.rs

@@ -1094,6 +1094,72 @@ pub unsafe extern "C" fn client_tokens(
     Result::Ok(tokens.into())
 }
 
+/// Subscribes to token updates
+///
+/// # Parameters
+/// * `client` - Pointer to ToriiClient instance
+/// * `contract_addresses` - Array of contract addresses
+/// * `callback` - Function called when updates occur
+///
+/// # Returns
+/// Result containing pointer to Subscription or error
+#[no_mangle]
+pub unsafe extern "C" fn client_on_token_update(
+    client: *mut ToriiClient,
+    contract_addresses: *const types::FieldElement,
+    contract_addresses_len: usize,
+    callback: unsafe extern "C" fn(Token),
+) -> Result<*mut Subscription> {
+    let client = Arc::new(unsafe { &*client });
+
+    // Convert contract addresses array to Vec<Felt> if not empty
+    let contract_addresses = if contract_addresses.is_null() || contract_addresses_len == 0 {
+        Vec::new()
+    } else {
+        let addresses =
+            unsafe { std::slice::from_raw_parts(contract_addresses, contract_addresses_len) };
+        addresses.iter().map(|f| (&f.clone()).into()).collect::<Vec<Felt>>()
+    };
+
+    let subscription_id = Arc::new(AtomicU64::new(0));
+    let (trigger, tripwire) = Tripwire::new();
+
+    let subscription = Subscription { id: Arc::clone(&subscription_id), trigger };
+
+    // Spawn a new thread to handle the stream and reconnections
+    let client_clone = client.clone();
+    RUNTIME.spawn(async move {
+        let mut backoff = Duration::from_secs(1);
+        let max_backoff = Duration::from_secs(60);
+
+        loop {
+            let rcv = client_clone.inner.on_token_updated(contract_addresses.clone()).await;
+
+            if let Ok(rcv) = rcv {
+                backoff = Duration::from_secs(1); // Reset backoff on successful connection
+
+                let mut rcv = rcv.take_until_if(tripwire.clone());
+
+                while let Some(Ok((id, token))) = rcv.next().await {
+                    subscription_id.store(id, Ordering::SeqCst);
+                    let token: Token = (&token).into();
+                    callback(token);
+                }
+            }
+
+            // If we've reached this point, the stream has ended (possibly due to disconnection)
+            // We'll try to reconnect after a delay, unless the tripwire has been triggered
+            if tripwire.clone().now_or_never().unwrap_or_default() {
+                break; // Exit the loop if the subscription has been cancelled
+            }
+            sleep(backoff).await;
+            backoff = std::cmp::min(backoff * 2, max_backoff);
+        }
+    });
+
+    Result::Ok(Box::into_raw(Box::new(subscription)))
+}
+
 /// Gets token balances for given accounts and contracts
 ///
 /// # Parameters

src/c/types.rs

@@ -103,6 +103,7 @@ impl From<&torii_grpc::types::Controller> for Controller {
 #[derive(Debug, Clone)]
 #[repr(C)]
 pub struct Token {
+    id: *const c_char,
     pub contract_address: FieldElement,
     pub name: *const c_char,
     pub symbol: *const c_char,
@@ -113,6 +114,7 @@ pub struct Token {
 impl From<&torii_grpc::types::Token> for Token {
     fn from(val: &torii_grpc::types::Token) -> Self {
         Token {
+            id: CString::new(val.id.clone()).unwrap().into_raw(),
             contract_address: (&val.contract_address).into(),
             name: CString::new(val.name.clone()).unwrap().into_raw(),
             symbol: CString::new(val.symbol.clone()).unwrap().into_raw(),

src/wasm/mod.rs

@@ -640,6 +640,74 @@ impl ToriiClient {
         Ok(Tokens(tokens.iter().map(|t| t.into()).collect()))
     }
 
+    /// Subscribes to token updates
+    ///
+    /// # Parameters
+    /// * `contract_addresses` - Array of contract addresses as hex strings
+    /// * `callback` - JavaScript function to call on updates
+    ///
+    /// # Returns
+    /// Result containing subscription handle or error
+    #[wasm_bindgen(js_name = onTokenUpdated)]
+    pub fn on_token_updated(
+        &self,
+        contract_addresses: Vec<String>,
+        callback: js_sys::Function,
+    ) -> Result<Subscription, JsValue> {
+        #[cfg(feature = "console-error-panic")]
+        console_error_panic_hook::set_once();
+
+        let contract_addresses = contract_addresses
+            .into_iter()
+            .map(|addr| {
+                Felt::from_str(&addr).map_err(|err| {
+                    JsValue::from(format!("failed to parse contract address: {err}"))
+                })
+            })
+            .collect::<Result<Vec<_>, _>>()?;
+
+        let subscription_id = Arc::new(AtomicU64::new(0));
+        let (trigger, tripwire) = Tripwire::new();
+
+        let subscription = Subscription { id: Arc::clone(&subscription_id), trigger };
+
+        // Spawn a new task to handle the stream and reconnections
+        let client = self.inner.clone();
+        let subscription_id_clone = Arc::clone(&subscription_id);
+        wasm_bindgen_futures::spawn_local(async move {
+            let mut backoff = 1000;
+            let max_backoff = 60000;
+
+            loop {
+                if let Ok(stream) = client.on_token_updated(contract_addresses.clone()).await {
+                    backoff = 1000; // Reset backoff on successful connection
+
+                    let mut stream = stream.take_until_if(tripwire.clone());
+
+                    while let Some(Ok((id, token))) = stream.next().await {
+                        subscription_id_clone.store(id, Ordering::SeqCst);
+                        let token: Token = (&token).into();
+
+                        let _ = callback.call1(
+                            &JsValue::null(),
+                            &token.serialize(&JSON_COMPAT_SERIALIZER).unwrap(),
+                        );
+                    }
+                }
+
+                // If we've reached this point, the stream has ended (possibly due to disconnection)
+                // We'll try to reconnect after a delay, unless the tripwire has been triggered
+                if tripwire.clone().now_or_never().unwrap_or_default() {
+                    break; // Exit the loop if the subscription has been cancelled
+                }
+                gloo_timers::future::TimeoutFuture::new(backoff).await;
+                backoff = std::cmp::min(backoff * 2, max_backoff);
+            }
+        });
+
+        Ok(subscription)
+    }
+
     /// Gets token balances for given accounts and contracts
     ///
     /// # Parameters

src/wasm/types.rs

@@ -46,6 +46,7 @@ pub struct TokenBalances(pub Vec<TokenBalance>);
 #[derive(Tsify, Serialize, Deserialize, Debug)]
 #[tsify(into_wasm_abi, from_wasm_abi)]
 pub struct Token {
+    pub id: String,
     pub contract_address: String,
     pub name: String,
     pub symbol: String,
@@ -56,6 +57,7 @@ pub struct Token {
 impl From<&torii_grpc::types::Token> for Token {
     fn from(value: &torii_grpc::types::Token) -> Self {
         Self {
+            id: value.id.clone(),
             contract_address: format!("{:#x}", value.contract_address),
             name: value.name.clone(),
             symbol: value.symbol.clone(),