fetch contract firsdt

crates/torii/core/src/executor.rs

@@ -208,18 +208,14 @@ impl<'c> Executor<'c> {
                     format!("Failed to execute query: {:?}, args: {:?}", statement, arguments)
                 })?;
 
-                sqlx::query("UPDATE contracts SET tps = ? WHERE id = ?")
+                let row = sqlx::query("UPDATE contracts SET tps = ? WHERE id = ? RETURNING *")
                     .bind(tps as i64)
                     .bind(format!("{:#x}", set_head.contract_address))
-                    .execute(&mut **tx)
+                    .fetch_one(&mut **tx)
                     .await?;
 
-                self.publish_queue.push(BrokerMessage::SetHead(ContractUpdated {
-                    head: set_head.head,
-                    tps,
-                    last_block_timestamp: set_head.last_block_timestamp,
-                    contract_address: set_head.contract_address,
-                }));
+                let contract = ContractUpdated::from_row(&row)?;
+                self.publish_queue.push(BrokerMessage::SetHead(contract));
             }
             QueryType::SetEntity(entity) => {
                 let row = query.fetch_one(&mut **tx).await.with_context(|| {

crates/torii/core/src/types.rs

@@ -88,8 +88,8 @@ pub struct Event {
 #[derive(FromRow, Deserialize, Debug, Clone)]
 #[serde(rename_all = "camelCase")]
 pub struct Contract {
-    pub head: u64,
-    pub tps: u64,
-    pub last_block_timestamp: u64,
-    pub contract_address: Felt,
+    pub head: i64,
+    pub tps: i64,
+    pub last_block_timestamp: i64,
+    pub contract_address: String,
 }

crates/torii/grpc/proto/world.proto

@@ -48,9 +48,9 @@ message SubscribeIndexerRequest {
 
 // A response containing indexer updates.
 message SubscribeIndexerResponse {
-    uint64 head = 1;
-    uint64 tps = 2;
-    uint64 last_block_timestamp = 3;
+    int64 head = 1;
+    int64 tps = 2;
+    int64 last_block_timestamp = 3;
     bytes contract_address = 4;
 }
 

crates/torii/grpc/src/server/mod.rs

@@ -696,7 +696,7 @@ impl DojoWorld {
         contract_address: Felt,
     ) -> Result<Receiver<Result<proto::world::SubscribeIndexerResponse, tonic::Status>>, Error>
     {
-        self.indexer_manager.add_subscriber(contract_address).await
+        self.indexer_manager.add_subscriber(&self.pool, contract_address).await
     }
 
     async fn subscribe_models(

crates/torii/grpc/src/server/subscriptions/indexer.rs

@@ -1,15 +1,17 @@
 use std::collections::HashMap;
 use std::future::Future;
 use std::pin::Pin;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::task::{Context, Poll};
 
 use futures::{Stream, StreamExt};
 use rand::Rng;
+use sqlx::{FromRow, Pool, Sqlite};
 use starknet::core::types::Felt;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
 use tokio::sync::RwLock;
-use torii_core::error::Error;
+use torii_core::error::{Error, ParseError};
 use torii_core::simple_broker::SimpleBroker;
 use torii_core::types::Contract as ContractUpdated;
 use tracing::{error, trace};
@@ -35,6 +37,7 @@ pub struct IndexerManager {
 impl IndexerManager {
     pub async fn add_subscriber(
         &self,
+        pool: &Pool<Sqlite>,
         contract_address: Felt,
     ) -> Result<Receiver<Result<proto::world::SubscribeIndexerResponse, tonic::Status>>, Error>
     {
@@ -44,11 +47,19 @@ impl IndexerManager {
         // NOTE: unlock issue with firefox/safari
         // initially send empty stream message to return from
         // initial subscribe call
+        let contract = sqlx::query(
+            "SELECT head, tps, last_block_timestamp, contract_address FROM contracts WHERE id = ?",
+        )
+        .bind(format!("{:#x}", contract_address))
+        .fetch_one(pool)
+        .await?;
+        let contract = ContractUpdated::from_row(&contract)?;
+
         let _ = sender
             .send(Ok(SubscribeIndexerResponse {
-                head: 0,
-                tps: 0,
-                last_block_timestamp: 0,
+                head: contract.head,
+                tps: contract.tps,
+                last_block_timestamp: contract.last_block_timestamp,
                 contract_address: contract_address.to_bytes_be().to_vec(),
             }))
             .await;
@@ -80,18 +91,19 @@ impl Service {
         update: &ContractUpdated,
     ) -> Result<(), Error> {
         let mut closed_stream = Vec::new();
+        let contract_address =
+            Felt::from_str(&update.contract_address).map_err(ParseError::FromStr)?;
 
         for (idx, sub) in subs.subscribers.read().await.iter() {
-            if sub.contract_address != Felt::ZERO && sub.contract_address != update.contract_address
-            {
+            if sub.contract_address != Felt::ZERO && sub.contract_address != contract_address {
                 continue;
             }
 
             let resp = SubscribeIndexerResponse {
                 head: update.head,
                 tps: update.tps,
                 last_block_timestamp: update.last_block_timestamp,
-                contract_address: update.contract_address.to_bytes_be().to_vec(),
+                contract_address: contract_address.to_bytes_be().to_vec(),
             };
 
             if sub.sender.send(Ok(resp)).await.is_err() {

crates/torii/grpc/src/types/mod.rs

@@ -18,9 +18,9 @@ pub mod schema;
 
 #[derive(Debug, Serialize, Deserialize, PartialEq, Hash, Eq, Clone)]
 pub struct IndexerUpdate {
-    pub head: u64,
-    pub tps: u64,
-    pub last_block_timestamp: u64,
+    pub head: i64,
+    pub tps: i64,
+    pub last_block_timestamp: i64,
     pub contract_address: Felt,
 }