Add customer swap attribution and volume reporting

migrations/20260324000000_create_issued_swap_calldata.sql

@@ -0,0 +1,28 @@
+CREATE TABLE IF NOT EXISTS issued_swap_calldata (
+    id               INTEGER PRIMARY KEY AUTOINCREMENT,
+    api_key_id       INTEGER NOT NULL,
+    key_id           TEXT    NOT NULL,
+    label            TEXT    NOT NULL,
+    owner            TEXT    NOT NULL,
+    chain_id         INTEGER NOT NULL,
+    taker            TEXT    NOT NULL,
+    to_address       TEXT    NOT NULL,
+    tx_value         TEXT    NOT NULL,
+    calldata         TEXT    NOT NULL,
+    calldata_hash    TEXT    NOT NULL,
+    input_token      TEXT    NOT NULL,
+    output_token     TEXT    NOT NULL,
+    output_amount    TEXT    NOT NULL,
+    maximum_io_ratio TEXT    NOT NULL,
+    estimated_input  TEXT    NOT NULL,
+    created_at       TEXT    NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_issued_swap_calldata_api_key_id_created
+    ON issued_swap_calldata (api_key_id, created_at);
+CREATE INDEX idx_issued_swap_calldata_chain_id_created
+    ON issued_swap_calldata (chain_id, created_at);
+CREATE INDEX idx_issued_swap_calldata_calldata_hash_created
+    ON issued_swap_calldata (calldata_hash, created_at);
+CREATE INDEX idx_issued_swap_calldata_taker_created
+    ON issued_swap_calldata (taker, created_at);

src/cli.rs

@@ -27,6 +27,13 @@ pub enum Command {
         #[command(subcommand)]
         command: KeysCommand,
     },
+    #[command(about = "Run internal reports")]
+    Report {
+        #[arg(long)]
+        config: PathBuf,
+        #[command(subcommand)]
+        command: ReportCommand,
+    },
 }
 
 #[derive(Subcommand)]
@@ -48,12 +55,26 @@ pub enum KeysCommand {
     Delete { key_id: String },
 }
 
+#[derive(Subcommand)]
+pub enum ReportCommand {
+    #[command(about = "Report executed customer swap volume")]
+    CustomerVolume {
+        #[arg(long)]
+        start_time: u64,
+        #[arg(long)]
+        end_time: u64,
+        #[arg(long, default_value_t = false)]
+        json: bool,
+    },
+}
+
 pub fn print_usage() {
     println!("Usage: st0x_rest_api <command>");
     println!();
     println!("Commands:");
     println!("  serve    Start the API server");
     println!("  keys     Manage API keys");
+    println!("  report   Run internal reports");
     println!();
     println!("Run 'st0x_rest_api <command> --help' for more information on a command.");
 }
@@ -74,6 +95,32 @@ pub async fn handle_keys_command(
     }
 }
 
+pub async fn handle_report_command(
+    command: ReportCommand,
+    pool: DbPool,
+    raindex: &crate::raindex::RaindexProvider,
+) -> Result<(), Box<dyn std::error::Error>> {
+    match command {
+        ReportCommand::CustomerVolume {
+            start_time,
+            end_time,
+            json,
+        } => {
+            crate::reporting::customer_volume::run(
+                &pool,
+                raindex,
+                crate::reporting::customer_volume::CustomerVolumeReportArgs {
+                    start_time,
+                    end_time,
+                    json,
+                },
+            )
+            .await?;
+            Ok(())
+        }
+    }
+}
+
 async fn create_key(
     pool: &DbPool,
     label: &str,
@@ -237,6 +284,40 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_report_parses_config_flag() {
+        let cli = Cli::try_parse_from([
+            "app",
+            "report",
+            "--config",
+            "/path/to/config.toml",
+            "customer-volume",
+            "--start-time",
+            "10",
+            "--end-time",
+            "20",
+        ])
+        .expect("parse");
+
+        match cli.command {
+            Some(Command::Report {
+                config,
+                command:
+                    ReportCommand::CustomerVolume {
+                        start_time,
+                        end_time,
+                        json,
+                    },
+            }) => {
+                assert_eq!(config, PathBuf::from("/path/to/config.toml"));
+                assert_eq!(start_time, 10);
+                assert_eq!(end_time, 20);
+                assert!(!json);
+            }
+            _ => panic!("expected Report command"),
+        }
+    }
+
     #[tokio::test]
     async fn test_create_key_inserts_row() {
         let pool = test_pool().await;

src/db/issued_swap_calldata.rs

@@ -0,0 +1,63 @@
+use crate::db::DbPool;
+
+pub(crate) struct NewIssuedSwapCalldata {
+    pub api_key_id: i64,
+    pub key_id: String,
+    pub label: String,
+    pub owner: String,
+    pub chain_id: i64,
+    pub taker: String,
+    pub to_address: String,
+    pub tx_value: String,
+    pub calldata: String,
+    pub calldata_hash: String,
+    pub input_token: String,
+    pub output_token: String,
+    pub output_amount: String,
+    pub maximum_io_ratio: String,
+    pub estimated_input: String,
+}
+
+pub(crate) async fn insert(
+    pool: &DbPool,
+    record: &NewIssuedSwapCalldata,
+) -> Result<(), sqlx::Error> {
+    sqlx::query(
+        "INSERT INTO issued_swap_calldata (
+            api_key_id,
+            key_id,
+            label,
+            owner,
+            chain_id,
+            taker,
+            to_address,
+            tx_value,
+            calldata,
+            calldata_hash,
+            input_token,
+            output_token,
+            output_amount,
+            maximum_io_ratio,
+            estimated_input
+        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+    )
+    .bind(record.api_key_id)
+    .bind(&record.key_id)
+    .bind(&record.label)
+    .bind(&record.owner)
+    .bind(record.chain_id)
+    .bind(&record.taker)
+    .bind(&record.to_address)
+    .bind(&record.tx_value)
+    .bind(&record.calldata)
+    .bind(&record.calldata_hash)
+    .bind(&record.input_token)
+    .bind(&record.output_token)
+    .bind(&record.output_amount)
+    .bind(&record.maximum_io_ratio)
+    .bind(&record.estimated_input)
+    .execute(pool)
+    .await?;
+
+    Ok(())
+}

src/db/mod.rs

@@ -1,3 +1,4 @@
+pub(crate) mod issued_swap_calldata;
 mod migrate;
 mod pool;
 pub(crate) mod settings;

src/main.rs

@@ -9,6 +9,7 @@ mod db;
 mod error;
 mod fairings;
 mod raindex;
+mod reporting;
 mod routes;
 mod telemetry;
 mod types;
@@ -110,6 +111,47 @@ fn configure_cors() -> Result<rocket_cors::Cors, StartupError> {
     .to_cors()?)
 }
 
+async fn load_raindex_provider(
+    cfg: &config::Config,
+    pool: &db::DbPool,
+) -> Result<raindex::RaindexProvider, String> {
+    let db_url = db::settings::get_setting(pool, "registry_url")
+        .await
+        .map_err(|e| format!("failed to read registry_url from database: {e}"))?
+        .filter(|url| !url.is_empty());
+
+    let registry_url = match db_url {
+        Some(url) => {
+            tracing::info!(registry_url = %url, "loaded registry_url from database");
+            url
+        }
+        None if !cfg.registry_url.is_empty() => {
+            if let Err(e) = db::settings::set_setting(pool, "registry_url", &cfg.registry_url).await
+            {
+                tracing::warn!(error = %e, "failed to seed registry_url into database");
+            }
+            cfg.registry_url.clone()
+        }
+        None => {
+            return Err("registry_url not found in database and not set in config file".into());
+        }
+    };
+
+    let local_db_path = std::path::PathBuf::from(&cfg.local_db_path);
+    if let Some(parent) = local_db_path.parent() {
+        std::fs::create_dir_all(parent).map_err(|e| {
+            format!(
+                "failed to create local db directory {}: {e}",
+                parent.display()
+            )
+        })?;
+    }
+
+    raindex::RaindexProvider::load(&registry_url, Some(local_db_path))
+        .await
+        .map_err(|e| format!("failed to load raindex registry: {e}"))
+}
+
 pub(crate) fn rocket(
     pool: db::DbPool,
     rate_limiter: fairings::RateLimiter,
@@ -159,7 +201,9 @@ async fn main() {
     };
 
     let config_path = match &command {
-        cli::Command::Serve { config } | cli::Command::Keys { config, .. } => config.clone(),
+        cli::Command::Serve { config }
+        | cli::Command::Keys { config, .. }
+        | cli::Command::Report { config, .. } => config.clone(),
     };
 
     let cfg = match config::Config::load(&config_path) {
@@ -195,56 +239,13 @@ async fn main() {
 
     match command {
         cli::Command::Serve { .. } => {
-            let db_url = db::settings::get_setting(&pool, "registry_url")
-                .await
-                .ok()
-                .flatten();
-
-            let registry_url = match db_url {
-                Some(url) if !url.is_empty() => {
-                    tracing::info!(registry_url = %url, "loaded registry_url from database");
-                    url
-                }
-                _ if !cfg.registry_url.is_empty() => {
-                    if let Err(e) =
-                        db::settings::set_setting(&pool, "registry_url", &cfg.registry_url).await
-                    {
-                        tracing::warn!(error = %e, "failed to seed registry_url into database");
-                    }
-                    cfg.registry_url
-                }
-                _ => {
-                    tracing::error!(
-                        "registry_url not found in database and not set in config file"
-                    );
-                    drop(log_guard);
-                    std::process::exit(1);
-                }
-            };
-
-            let local_db_path = std::path::PathBuf::from(&cfg.local_db_path);
-            if let Some(parent) = local_db_path.parent() {
-                if !parent.exists() {
-                    if let Err(e) = std::fs::create_dir_all(parent) {
-                        tracing::error!(error = %e, path = %parent.display(), "failed to create local db directory");
-                        drop(log_guard);
-                        std::process::exit(1);
-                    }
-                }
-            }
-
-            let raindex_config = match raindex::RaindexProvider::load(
-                &registry_url,
-                Some(local_db_path),
-            )
-            .await
-            {
+            let raindex_config = match load_raindex_provider(&cfg, &pool).await {
                 Ok(config) => {
-                    tracing::info!(registry_url = %registry_url, "raindex registry loaded");
+                    tracing::info!("raindex registry loaded");
                     config
                 }
                 Err(e) => {
-                    tracing::error!(error = %e, registry_url = %registry_url, "failed to load raindex registry");
+                    tracing::error!(error = %e, "failed to load raindex registry");
                     drop(log_guard);
                     std::process::exit(1);
                 }
@@ -283,6 +284,25 @@ async fn main() {
                 std::process::exit(1);
             }
         }
+        cli::Command::Report { command, .. } => {
+            let raindex_config = match load_raindex_provider(&cfg, &pool).await {
+                Ok(config) => {
+                    tracing::info!("raindex registry loaded for report");
+                    config
+                }
+                Err(e) => {
+                    tracing::error!(error = %e, "failed to load raindex registry");
+                    drop(log_guard);
+                    std::process::exit(1);
+                }
+            };
+
+            if let Err(e) = cli::handle_report_command(command, pool, &raindex_config).await {
+                tracing::error!(error = %e, "report command failed");
+                drop(log_guard);
+                std::process::exit(1);
+            }
+        }
     }
 
     drop(log_guard);