diff --git a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go index cef588c4..a912795e 100644 --- a/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go +++ b/harnesses/aggregator-head-lag/cmd/script/head_lag_monitor.go @@ -4,6 +4,8 @@ import ( "encoding/json" "fmt" "log" + "net/http" + "net/url" "strings" "sync" "time" @@ -419,11 +421,19 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error { return fmt.Errorf("failed to get JWT token: %w", err) } - log.Printf("[HEAD-LAG][CODEX] Step 2/4: Creating proxy dialer...") - dialer := getProxyDialerWithSubprotocols([]string{"graphql-transport-ws"}) + log.Printf("[HEAD-LAG][CODEX] Step 2/4: Creating direct dialer (no proxy — IP must match JWE origin)...") + dialer := &websocket.Dialer{ + Subprotocols: []string{"graphql-transport-ws"}, + HandshakeTimeout: 30 * time.Second, + } log.Printf("[HEAD-LAG][CODEX] Step 3/4: Connecting to wss://graph.codex.io/graphql...") - conn, resp, err := dialer.Dial("wss://graph.codex.io/graphql", nil) + cookieVal := url.QueryEscape(`{"token":"` + jwtToken + `"}`) + wsHeaders := http.Header{} + wsHeaders.Set("Cookie", "codex_token="+cookieVal) + wsHeaders.Set("Origin", "https://www.defined.fi") + wsHeaders.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36") + conn, resp, err := dialer.Dial("wss://graph.codex.io/graphql", wsHeaders) if err != nil { if resp != nil { return fmt.Errorf("dial failed (HTTP %d): %w", resp.StatusCode, err) @@ -439,12 +449,11 @@ func connectAndMonitorCodex(config *Config, stopChan <-chan struct{}) error { log.Printf("[HEAD-LAG][CODEX] Step 3/4: ✅ WebSocket connection established (IP check failed: %v)", err) } - // Connection init with JWT Bearer token - log.Printf("[HEAD-LAG][CODEX] Step 4/4: Sending connection_init with JWT...") + log.Printf("[HEAD-LAG][CODEX] Step 4/4: Sending connection_init (token len=%d)...", len(jwtToken)) initMsg := map[string]interface{}{ "type": "connection_init", "payload": map[string]interface{}{ - "Authorization": fmt.Sprintf("Bearer %s", jwtToken), + "Authorization": "Bearer " + jwtToken, }, } if err := conn.WriteJSON(initMsg); err != nil { diff --git a/harnesses/solana-exec/cmd/api/main.go b/harnesses/solana-exec/cmd/api/main.go index 9cecff45..3fcf7cac 100644 --- a/harnesses/solana-exec/cmd/api/main.go +++ b/harnesses/solana-exec/cmd/api/main.go @@ -64,29 +64,42 @@ func handleExecLeaderboard(pool *pgxpool.Pool) http.HandlerFunc { rows, err := pool.Query(ctx, ` SELECT platform, - AVG(avg_priority_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_prio, - AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p50, - AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p95, - AVG(avg_platform_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_pfee, - AVG(jito_rate) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_jito, - AVG(avg_cu_consumed) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_cu, - SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_count, - - AVG(avg_priority_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_prio, - AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p50, - AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p95, - AVG(avg_platform_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_pfee, - AVG(jito_rate) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_jito, - AVG(avg_cu_consumed) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_cu, - SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_count, - - AVG(avg_priority_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_prio, - AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p50, - AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p95, - AVG(avg_platform_fee_lamports) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_pfee, - AVG(jito_rate) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_jito, - AVG(avg_cu_consumed) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_cu, - SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_count, + -- weighted averages: SUM(avg×count)/SUM(count) avoids skewing by small off-peak buckets + SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_prio, + AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p50, + AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_p95, + SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_pfee, + SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_jito, + SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_cu, + SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '24 hours') AS h24_count, + + SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_prio, + AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p50, + AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_p95, + SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_pfee, + SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_jito, + SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_cu, + SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '7 days') AS d7_count, + + SUM(avg_priority_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_prio, + AVG(p50_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p50, + AVG(p95_cu_price_micro) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_p95, + SUM(avg_platform_fee_lamports * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_pfee, + SUM(jito_rate * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_jito, + SUM(avg_cu_consumed * tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') + / NULLIF(SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_cu, + SUM(tx_count) FILTER (WHERE bucket_start >= now() - INTERVAL '30 days') AS d30_count, MAX(bucket_start)::text AS latest_bucket FROM solana_exec_facts