(rpc_server) fix: fetch state by pages (#364)

* fix fetch state by pages

* fmt

* changelog

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### What's Changed
 * Improved bulk insertion of state_changes, reducing database requests from hundreds to a maximum of 7 per block.
 * Configuration improvement. Create default config.toml on start application to loaded parameters from the environment variables.
+* Fix to fetch state by pages (view_state_paginated).
 
 ## [0.3.0](https://github.com/near/read-rpc/releases/tag/v0.3.0)
 

database/src/postgres/mod.rs

@@ -9,22 +9,22 @@ static SHARD_DB_MIGRATOR: sqlx::migrate::Migrator =
 
 #[derive(borsh::BorshSerialize, borsh::BorshDeserialize, Clone, Debug)]
 struct PageState {
+    pub last_data_key: Option<String>,
     pub page_size: i64,
-    pub offset: i64,
 }
 
 impl PageState {
     fn new(page_size: i64) -> Self {
         Self {
+            last_data_key: None,
             page_size,
-            offset: 0,
         }
     }
 
-    fn next_page(&self) -> Self {
+    fn next_page(&self, last_data_key: String) -> Self {
         Self {
+            last_data_key: Some(last_data_key),
             page_size: self.page_size,
-            offset: self.offset + self.page_size,
         }
     }
 }

database/src/postgres/rpc_server.rs

@@ -100,27 +100,35 @@ impl crate::ReaderDbManager for crate::PostgresDBManager {
                 WHERE
                     sc.account_id = $1
                     AND sc.data_value IS NOT NULL
+                    AND (
+                        $3 IS NULL OR
+                        sc.data_key > $3
+                    )
                 ORDER BY 
-                    sc.data_key
-                LIMIT $3 OFFSET $4;
+                    sc.data_key ASC
+                LIMIT $4;
                 ",
         )
         .bind(account_id.to_string())
         .bind(bigdecimal::BigDecimal::from(block_height))
+        .bind(page_state.last_data_key.clone())
         .bind(page_state.page_size)
-        .bind(page_state.offset)
         .fetch(shard_id_pool.pool);
         let mut items = std::collections::HashMap::new();
+        let mut last_data_key = String::new();
         while let Some(row) = stream.next().await {
             let (key, value): (String, Vec<u8>) = row?;
+            last_data_key.clone_from(&key);
             items.insert(hex::decode(key)?, value);
         }
         if items.len() < page_state.page_size as usize {
             Ok((items, None))
         } else {
             Ok((
                 items,
-                Some(hex::encode(borsh::to_vec(&page_state.next_page())?)),
+                Some(hex::encode(borsh::to_vec(
+                    &page_state.next_page(last_data_key),
+                )?)),
             ))
         }
     }