[WIP] Add analytics for opportunity

Tiltfile

@@ -75,7 +75,7 @@ local_resource(
 
 local_resource(
     "auction-server",
-    serve_cmd="source ../tilt-resources.env; source ./.env; cargo run -- run --database-url $DATABASE_URL --secret-key $SECRET_KEY",
+    serve_cmd="cat ./.env; source ../tilt-resources.env; env $(grep -v '^#' .env | xargs) cargo run -- run",
     serve_dir="auction-server",
     resource_deps=["create-server-configs", "svm-build-programs", "svm-setup-accounts"],
     readiness_probe=probe(period_secs=5, http_get=http_get_action(port=9000)),

auction-server/.env.example

@@ -1 +1,6 @@
 DATABASE_URL=postgresql://postgres@localhost/postgres
+
+DATABASE_URL_CLICKHOUSE="http://localhost:18123"
+DATABASE_NAME_CLICKHOUSE="default"
+DATABASE_USER_CLICKHOUSE="default"
+DATABASE_PASSWORD_CLICKHOUSE="clickhouse"

auction-server/Cargo.toml

@@ -60,6 +60,7 @@ spl-memo-client = { workspace = true }
 spl-token-2022 = { workspace = true }
 humantime-serde = "1.1.1"
 tokio-metrics = { version = "0.4.2", features = ["rt"] }
+clickhouse = { version = "0.13.2", features = ["time", "uuid"] }
 
 [dev-dependencies]
 mockall = "0.13.1"

auction-server/clickhouse_migrations/0001_add_opportunity.sql

@@ -0,0 +1,52 @@
+CREATE TABLE IF NOT EXISTS opportunity (
+    id UUID,
+    creation_time DateTime64(6),
+    permission_key FixedString(64),
+    chain_id String,
+
+    program LowCardinality(String), -- "limit_order" or "quote"
+
+    sell_tokens String,
+    buy_tokens String,
+
+    -- Token info
+    sell_token_mint String,
+    sell_token_amount UInt64,
+    sell_token_usd_price Nullable(Float64),
+
+    buy_token_mint String,
+    buy_token_amount UInt64,
+    buy_token_usd_price Nullable(Float64),
+
+    -- Optional removal tracking
+    removal_time Nullable(DateTime64(6)),
+    removal_reason Nullable(String),
+
+    -- === Flattened Metadata Fields ===
+
+    -- For Limo variant
+    limo_order Nullable(String),
+    limo_order_address Nullable(String),
+    limo_slot Nullable(UInt64),
+
+    -- For Swap variant
+    swap_user_wallet_address Nullable(String),
+    swap_fee_token Nullable(String),
+    swap_referral_fee_bps Nullable(UInt16),
+    swap_referral_fee_ppm Nullable(UInt64),
+    swap_platform_fee_bps Nullable(UInt64),
+    swap_platform_fee_ppm Nullable(UInt64),
+    swap_token_program_user Nullable(String),
+    swap_token_program_searcher Nullable(String),
+    swap_token_account_initialization_configs Nullable(String), -- store as JSON
+    swap_user_mint_user_balance Nullable(UInt64),
+    swap_memo Nullable(String),
+    swap_cancellable Nullable(Bool),
+    swap_minimum_lifetime Nullable(UInt32),
+
+    -- Profile
+    profile_id Nullable(UUID),
+    profile_name Nullable(String)
+
+) ENGINE = ReplacingMergeTree
+ORDER BY (creation_time, id);

auction-server/src/config.rs

@@ -12,6 +12,7 @@ use {
         serde_as,
         DisplayFromStr,
     },
+    server::ClickhouseConfig,
     solana_sdk::pubkey::Pubkey,
     std::{
         collections::HashMap,
@@ -20,7 +21,7 @@ use {
     },
 };
 
-mod server;
+pub mod server;
 
 // `Options` is a structup definition to provide clean command-line args for Hermes.
 #[derive(Parser, Debug)]
@@ -34,6 +35,8 @@ pub enum Options {
     Run(RunOptions),
     /// Run db migrations and exit.
     Migrate(MigrateOptions),
+    /// Run clickhouse migrations and exit.
+    MigrateClickhouse(ClickhouseConfig),
 }
 
 #[derive(Args, Clone, Debug)]

auction-server/src/config/server.rs

@@ -8,6 +8,27 @@ const DEFAULT_METRICS_ADDR: &str = "127.0.0.1:9001";
 const DEFAULT_DATABASE_CONNECTIONS: &str = "10";
 const DEFAULT_REQUESTER_IP_HEADER_NAME: &str = "X-Forwarded-For";
 
+#[derive(Args, Clone, Debug)]
+pub struct ClickhouseConfig {
+    /// database url to run the migrations for.
+    #[arg(long = "database-url-clickhouse")]
+    #[arg(env = "DATABASE_URL_CLICKHOUSE")]
+    pub database_url_clickhouse:      String,
+    /// database name to run the migrations for.
+    #[arg(long = "database-name-clickhouse")]
+    #[arg(env = "DATABASE_NAME_CLICKHOUSE")]
+    pub database_name_clickhouse:     String,
+    /// database user to run the migrations for.
+    #[arg(long = "database-user-clickhouse")]
+    #[arg(env = "DATABASE_USER_CLICKHOUSE")]
+    pub database_user_clickhouse:     String,
+    /// database password to run the migrations for.
+    #[arg(long = "database-password-clickhouse")]
+    #[arg(env = "DATABASE_PASSWORD_CLICKHOUSE")]
+    pub database_password_clickhouse: String,
+}
+
+
 #[derive(Args, Clone, Debug)]
 #[command(next_help_heading = "Server Options")]
 #[group(id = "Server")]
@@ -41,4 +62,7 @@ pub struct Options {
     #[arg(default_value = DEFAULT_REQUESTER_IP_HEADER_NAME)]
     #[arg(env = "REQUESTER_IP_HEADER_NAME")]
     pub requester_ip_header_name: String,
+    /// Clickhouse database config
+    #[command(flatten)]
+    pub clickhouse_config:        ClickhouseConfig,
 }

auction-server/src/kernel/db.rs

@@ -4,3 +4,4 @@ use sqlx::{
 };
 
 pub type DB = Pool<Postgres>;
+pub type DBAnalytics = clickhouse::Client;

auction-server/src/main.rs

@@ -1,7 +1,6 @@
 #![cfg_attr(test, allow(dead_code))]
 
 use {
-    crate::server::run_migrations,
     anyhow::Result,
     clap::Parser,
     opentelemetry::KeyValue,
@@ -14,7 +13,11 @@ use {
         is_metrics,
         MetricsLayer,
     },
-    server::start_server,
+    server::{
+        run_migrations,
+        run_migrations_clichouse,
+        start_server,
+    },
     std::{
         io::IsTerminal,
         time::Duration,
@@ -107,5 +110,6 @@ async fn main() -> Result<()> {
     match config::Options::parse() {
         config::Options::Run(opts) => start_server(opts).await,
         config::Options::Migrate(opts) => run_migrations(opts).await,
+        config::Options::MigrateClickhouse(opts) => run_migrations_clichouse(opts).await,
     }
 }

auction-server/src/opportunity/repository/add_opportunity_analytics.rs

@@ -0,0 +1,18 @@
+use {
+    super::Repository,
+    crate::{
+        api::RestError,
+        opportunity::entities::OpportunitySvm,
+    },
+};
+
+impl Repository {
+    pub async fn add_opportunity_analytics(
+        &self,
+        opportunity: OpportunitySvm,
+    ) -> Result<(), RestError> {
+        self.db_analytics
+            .add_opportunity(&opportunity, None, None)
+            .await
+    }
+}

auction-server/src/opportunity/repository/mod.rs

@@ -11,6 +11,7 @@ use {
 };
 
 mod add_opportunity;
+mod add_opportunity_analytics;
 mod get_express_relay_metadata;
 mod get_in_memory_opportunities;
 mod get_in_memory_opportunities_by_key;
@@ -29,6 +30,7 @@ pub const OPPORTUNITY_PAGE_SIZE_CAP: usize = 100;
 pub struct Repository {
     pub in_memory_store: InMemoryStoreSvm,
     pub db:              Box<dyn Database>,
+    pub db_analytics:    Box<dyn DatabaseAnalytics>,
 }
 
 
@@ -70,10 +72,11 @@ impl Deref for InMemoryStoreSvm {
 }
 
 impl Repository {
-    pub fn new(db: impl Database) -> Self {
+    pub fn new(db: impl Database, db_analytics: impl DatabaseAnalytics) -> Self {
         Self {
             in_memory_store: InMemoryStoreSvm::new(),
             db:              Box::new(db),
+            db_analytics:    Box::new(db_analytics),
         }
     }
     pub(super) async fn update_metrics(&self) {