rpc: Unique addresses

RPC method to show number of unique addresses interracting with cauldron

src/db/bcmr/mod.rs

@@ -15,6 +15,7 @@ use rusqlite::{params, Connection};
 const MAX_DOWNLOAD_ATTEMPTS: usize = 100;
 
 // BCMR auth chain entry, matches auth_chain_entry table
+#[allow(dead_code)]
 pub struct AuthChainEntry {
     pub utxo: OutPointHash,
     pub txid: Txid,

src/db/cauldron/mod.rs

@@ -13,6 +13,7 @@ pub mod header;
 pub mod mempool;
 pub mod pool;
 pub mod tx;
+pub mod user;
 pub mod utxo_funding;
 pub mod utxo_spending;
 
@@ -20,6 +21,7 @@ pub fn prepare_tables(conn: &Connection) {
     tx::create_table(conn);
     utxo_funding::create_table(conn);
     utxo_spending::create_table(conn);
+    user::create_table(conn);
 
     conn.execute(
         "CREATE TABLE config (

src/db/cauldron/user.rs

@@ -0,0 +1,107 @@
+// Copyright (C) 2024 Riften Labs AS
+//
+// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
+// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
+
+// Keep track of individual user activity
+
+use std::collections::HashSet;
+
+use anyhow::Result;
+use bitcoin_hashes::{hex::ToHex, Hash};
+use bitcoincash::{PubkeyHash, Transaction};
+use riftenlabs_defi::cauldron::ParsedContract;
+use rusqlite::{params, Connection};
+
+pub fn create_table(conn: &Connection) {
+    conn.execute(
+        "CREATE TABLE user_action (
+            spent_utxo_hash TEXT NOT NULL,
+            user TEXT NOT NULL,
+            address_type TEXT NOT NULL,
+            FOREIGN KEY(spent_utxo_hash) REFERENCES utxo_spending(spent_utxo_hash),
+            PRIMARY KEY (spent_utxo_hash, user)
+        )",
+        [],
+    )
+    .unwrap();
+}
+
+pub fn insert_user_action(
+    con: &Connection,
+    cauldrons: &Vec<ParsedContract>,
+    tx: &Transaction,
+    replace: bool,
+) -> Result<()> {
+    let mut statement = con.prepare(&format!(
+        "INSERT OR {} INTO user_action (spent_utxo_hash, user, address_type) VALUES (?, ?, ?)",
+        if replace { "REPLACE" } else { "IGNORE" }
+    ))?;
+
+    let p2pkh_output_hashes: HashSet<PubkeyHash> = tx
+        .output
+        .iter()
+        .filter_map(|output| {
+            if output.script_pubkey.is_p2pkh() {
+                Some(output.script_pubkey[3..23].to_vec())
+            } else {
+                None
+            }
+        })
+        .map(|pubkey| {
+            PubkeyHash::from_inner(
+                pubkey
+                    .try_into()
+                    .expect("failed to convert into pubkeyhash"),
+            )
+        })
+        .collect();
+
+    for c in cauldrons {
+        for user in &p2pkh_output_hashes {
+            statement.execute(params![c.spent_utxo_hash.to_hex(), user.to_hex(), "p2pkh"])?;
+        }
+    }
+    Ok(())
+}
+
+pub fn get_unique_per_month_accumilating(conn: &Connection) -> Result<Vec<(String, usize)>> {
+    let sql = "WITH first_user_action AS (
+    -- Step 1: Get the first month when each user was active
+    SELECT
+        ua.user,
+        MIN(strftime('%Y-%m', datetime(tx.mtp_timestamp, 'unixepoch'))) AS first_action_month
+    FROM user_action ua
+    JOIN utxo_spending us ON ua.spent_utxo_hash = us.spent_utxo_hash
+    JOIN tx ON us.txid = tx.txid
+    GROUP BY ua.user
+),
+-- Step 2: Accumulate unique users by month
+cumulative_unique_users AS (
+    SELECT
+        f.first_action_month AS month_year,
+        COUNT(DISTINCT f.user) AS new_unique_users
+    FROM first_user_action f
+    GROUP BY f.first_action_month
+    ORDER BY f.first_action_month
+)
+-- Step 3: Running total of unique users
+SELECT
+    month_year,
+    SUM(new_unique_users) OVER (ORDER BY month_year) AS accumulated_unique_users
+FROM cumulative_unique_users;
+";
+
+    let mut stmt = conn.prepare(sql)?;
+    let mut rows = stmt.query([])?;
+    let mut results = Vec::new();
+
+    while let Some(row) = rows.next()? {
+        let month_year: String = row.get(0)?;
+        let accumulated_unique_users: usize = row.get(1)?;
+
+        results.push((month_year, accumulated_unique_users));
+    }
+
+    Ok(results)
+}

src/index.rs

@@ -24,6 +24,7 @@ use crate::{
             config::{config_get, config_set},
             header::{db_get_header, store_headers},
             tx::insert_block_tx,
+            user::insert_user_action,
             utxo_funding::insert_utxo_funding,
             utxo_spending::insert_utxo_spending,
         },
@@ -86,6 +87,7 @@ pub fn update_mempool(db: DBPool, electrum: Arc<Mutex<Client>>) -> Result<()> {
 
         insert_utxo_funding(&db_tx, &cauldrons, &txid, false)?;
         insert_utxo_spending(&db_tx, &cauldrons, &txid, false)?;
+        insert_user_action(&db_tx, &cauldrons, &tx, true)?;
 
         all_cauldrons.extend(cauldrons);
     }
@@ -226,6 +228,8 @@ pub fn index_blocks(
                 .context("inserting funding utxos")?;
             insert_utxo_spending(&db_tx, &cauldrons, &txid, true)
                 .context("inserting spending utxos")?;
+            insert_user_action(&db_tx, &cauldrons, tx, true).context("inserting user actions")?;
+
             total_cauldrons += cauldrons.len();
 
             all_cauldrons.extend(cauldrons);

src/main.rs

@@ -247,7 +247,8 @@ fn launch() -> _ {
                 rpc::pool::list_active_pools,
                 rpc::contract::contract_count_token,
                 rpc::contract::contract_count_all,
-                rpc::contract::contract_volume
+                rpc::contract::contract_volume,
+                rpc::user::unique_addresses
             ],
         )
         .mount("/bcmr", routes![rpc::bcmr::token_bcmr,])

src/rpc/mod.rs

@@ -19,6 +19,7 @@ pub mod pool;
 pub mod price;
 pub mod tokens;
 pub mod tvl;
+pub mod user;
 
 pub type ResponseCache = Arc<Mutex<HashMap<String, (i64 /* timestamp */, Vec<serde_json::Value>)>>>;
 

src/rpc/user.rs

@@ -0,0 +1,26 @@
+// Copyright (C) 2024 Riften Labs AS
+//
+// This software is licensed under the GNU Affero General Public License (AGPL), version 3.0 or later.
+// A copy of the license can be found in the LICENSE file or at https://www.gnu.org/licenses/agpl-3.0.html
+
+use rocket::get;
+use rocket::{http::Status, response::status::Custom, serde::json::Json, State};
+use serde_json::json;
+use serde_json::Value;
+
+use crate::db::{cauldron::user::get_unique_per_month_accumilating, DB};
+
+#[get("/user/unique_addresses")]
+pub fn unique_addresses(conn: &State<DB>) -> Result<Json<Value>, Custom<String>> {
+    let db = conn.cauldron_r.get().map_err(|_| {
+        Custom(
+            Status::InternalServerError,
+            "Failed to get DB connection".into(),
+        )
+    })?;
+
+    let users = get_unique_per_month_accumilating(&db)
+        .map_err(|e| Custom(Status::InternalServerError, format!("Error: {}", e)))?;
+
+    Ok(Json(json!(users)))
+}