DeepBook V3 TVL endpoint (#19962)

## Description 

Added an endpoint that calculates the net deposits at some timestamp.

## Test plan 

Tested locally.

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] Indexer: 
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:
- [ ] REST API:

crates/sui-deepbook-indexer/src/models.rs

@@ -2,14 +2,14 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use diesel::data_types::PgTimestamp;
-use diesel::{Identifiable, Insertable, Queryable, Selectable};
+use diesel::{Identifiable, Insertable, Queryable, QueryableByName, Selectable};
 
 use serde::Serialize;
 use sui_indexer_builder::{Task, LIVE_TASK_TARGET_CHECKPOINT};
 
 use crate::schema::{
-    balances, flashloans, order_fills, order_updates, pool_prices, pools, progress_store,
-    proposals, rebates, stakes, sui_error_transactions, trade_params_update, votes,
+    balances, balances_summary, flashloans, order_fills, order_updates, pool_prices, pools,
+    progress_store, proposals, rebates, stakes, sui_error_transactions, trade_params_update, votes,
 };
 
 #[derive(Queryable, Selectable, Insertable, Identifiable, Debug)]
@@ -65,6 +65,14 @@ pub struct OrderFillSummary {
     pub base_quantity: i64,
 }
 
+#[derive(QueryableByName, Debug, Serialize)]
+#[diesel(table_name = balances_summary)]
+pub struct BalancesSummary {
+    pub asset: String,
+    pub amount: i64,
+    pub deposit: bool,
+}
+
 #[derive(Queryable, Selectable, Insertable, Identifiable, Debug)]
 #[diesel(table_name = flashloans, primary_key(digest))]
 pub struct Flashloan {

crates/sui-deepbook-indexer/src/schema.rs

@@ -17,6 +17,14 @@ diesel::table! {
     }
 }
 
+diesel::table! {
+    balances_summary (asset) {
+        asset -> Text,
+        amount -> Int8,
+        deposit -> Bool,
+    }
+}
+
 diesel::table! {
     flashloans (id) {
         id -> Int4,

crates/sui-deepbook-indexer/src/server.rs

@@ -3,8 +3,8 @@
 
 use crate::{
     error::DeepBookError,
-    models::{OrderFillSummary, Pools},
-    schema,
+    models::{BalancesSummary, OrderFillSummary, Pools},
+    schema::{self},
     sui_deepbook_indexer::PgDeepbookPersistent,
 };
 use axum::{
@@ -18,14 +18,15 @@ use diesel::BoolExpressionMethods;
 use diesel::QueryDsl;
 use diesel::{ExpressionMethods, SelectableHelper};
 use diesel_async::RunQueryDsl;
-use std::net::SocketAddr;
 use std::time::{SystemTime, UNIX_EPOCH};
+use std::{collections::HashMap, net::SocketAddr};
 use tokio::{net::TcpListener, task::JoinHandle};
 
 pub const GET_POOLS_PATH: &str = "/get_pools";
 pub const GET_24HR_VOLUME_PATH: &str = "/get_24hr_volume/:pool_id";
 pub const GET_24HR_VOLUME_BY_BALANCE_MANAGER_ID: &str =
     "/get_24hr_volume_by_balance_manager_id/:pool_id/:balance_manager_id";
+pub const GET_NET_DEPOSITS: &str = "/get_net_deposits/:asset_ids/:timestamp";
 
 pub fn run_server(socket_address: SocketAddr, state: PgDeepbookPersistent) -> JoinHandle<()> {
     tokio::spawn(async move {
@@ -43,6 +44,7 @@ pub(crate) fn make_router(state: PgDeepbookPersistent) -> Router {
             GET_24HR_VOLUME_BY_BALANCE_MANAGER_ID,
             get(get_24hr_volume_by_balance_manager_id),
         )
+        .route(GET_NET_DEPOSITS, get(get_net_deposits))
         .with_state(state)
 }
 
@@ -142,3 +144,43 @@ async fn get_24hr_volume_by_balance_manager_id(
 
     Ok(Json(vec![maker_vol, taker_vol]))
 }
+
+#[debug_handler]
+async fn get_net_deposits(
+    Path((asset_ids, timestamp)): Path<(String, String)>,
+    State(state): State<PgDeepbookPersistent>,
+) -> Result<Json<HashMap<String, i64>>, DeepBookError> {
+    let connection = &mut state.pool.get().await?;
+    let mut query =
+        "SELECT asset, SUM(amount)::bigint AS amount, deposit FROM balances WHERE timestamp < to_timestamp("
+            .to_string();
+    query.push_str(&timestamp);
+    query.push_str(") AND asset in (");
+    for asset in asset_ids.split(",") {
+        if asset.starts_with("0x") {
+            let len = asset.len();
+            query.push_str(&format!("'{}',", &asset[2..len]));
+        } else {
+            query.push_str(&format!("'{}',", asset));
+        }
+    }
+    query.pop();
+    query.push_str(") GROUP BY asset, deposit");
+
+    let results: Vec<BalancesSummary> = diesel::sql_query(query).load(connection).await?;
+    let mut net_deposits = HashMap::new();
+    for result in results {
+        let mut asset = result.asset;
+        if !asset.starts_with("0x") {
+            asset.insert_str(0, "0x");
+        }
+        let amount = result.amount;
+        if result.deposit {
+            *net_deposits.entry(asset).or_insert(0) += amount;
+        } else {
+            *net_deposits.entry(asset).or_insert(0) -= amount;
+        }
+    }
+
+    Ok(Json(net_deposits))
+}