test: GitHub Actions codex token push

harnesses/solana-exec/cmd/api/main.go

@@ -62,48 +62,67 @@ func handleExecLeaderboard(pool *pgxpool.Pool) http.HandlerFunc {
 		defer cancel()
 
 		rows, err := pool.Query(ctx, `
+			WITH cu AS (
+				SELECT platform,
+					percentile_cont(0.5) WITHIN GROUP (ORDER BY cu_price_micro)
+						FILTER (WHERE block_time >= now() - INTERVAL '24 hours') AS h24_p50,
+					percentile_cont(0.95) WITHIN GROUP (ORDER BY cu_price_micro)
+						FILTER (WHERE block_time >= now() - INTERVAL '24 hours') AS h24_p95,
+					percentile_cont(0.5) WITHIN GROUP (ORDER BY cu_price_micro)
+						FILTER (WHERE block_time >= now() - INTERVAL '7 days') AS d7_p50,
+					percentile_cont(0.95) WITHIN GROUP (ORDER BY cu_price_micro)
+						FILTER (WHERE block_time >= now() - INTERVAL '7 days') AS d7_p95,
+					percentile_cont(0.5) WITHIN GROUP (ORDER BY cu_price_micro)
+						FILTER (WHERE block_time >= now() - INTERVAL '30 days') AS d30_p50,
+					percentile_cont(0.95) WITHIN GROUP (ORDER BY cu_price_micro)
+						FILTER (WHERE block_time >= now() - INTERVAL '30 days') AS d30_p95
+				FROM solana_exec_cu_samples
+				GROUP BY platform
+			)
 			SELECT
-				platform,
+				f.platform,
 				-- 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
-			GROUP BY platform`,
+				SUM(f.avg_priority_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_prio,
+				MAX(c.h24_p50) AS h24_p50,
+				MAX(c.h24_p95) AS h24_p95,
+				SUM(f.avg_platform_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_pfee,
+				SUM(f.jito_rate * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_jito,
+				SUM(f.avg_cu_consumed * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours'), 0) AS h24_cu,
+				SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '24 hours') AS h24_count,
+
+				SUM(f.avg_priority_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_prio,
+				MAX(c.d7_p50) AS d7_p50,
+				MAX(c.d7_p95) AS d7_p95,
+				SUM(f.avg_platform_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_pfee,
+				SUM(f.jito_rate * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_jito,
+				SUM(f.avg_cu_consumed * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days'), 0) AS d7_cu,
+				SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '7 days') AS d7_count,
+
+				SUM(f.avg_priority_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_prio,
+				MAX(c.d30_p50) AS d30_p50,
+				MAX(c.d30_p95) AS d30_p95,
+				SUM(f.avg_platform_fee_lamports * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_pfee,
+				SUM(f.jito_rate * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_jito,
+				SUM(f.avg_cu_consumed * f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days')
+				  / NULLIF(SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days'), 0) AS d30_cu,
+				SUM(f.tx_count) FILTER (WHERE f.bucket_start >= now() - INTERVAL '30 days') AS d30_count,
+
+				MAX(f.bucket_start)::text AS latest_bucket
+			FROM solana_exec_facts f
+			LEFT JOIN cu c USING (platform)
+			WHERE f.bucket_start >= now() - INTERVAL '31 days'
+			GROUP BY f.platform`,
 		)
 		if err != nil {
 			log.Printf("exec-api: query: %v", err)

harnesses/solana-exec/cmd/materializer/main.go

@@ -27,6 +27,12 @@ func main() {
 				log.Printf("materializer: %s: done", plt)
 			}
 		}
+		if err := db.PurgeCUSamples(ctx); err != nil {
+			log.Printf("materializer: purge cu samples: %v", err)
+		}
+		if err := db.PurgeEvents(ctx); err != nil {
+			log.Printf("materializer: purge events: %v", err)
+		}
 		time.Sleep(5 * time.Minute)
 	}
 }

harnesses/solana-exec/internal/helius/client.go

@@ -10,22 +10,32 @@ import (
 )
 
 // Client wraps the Helius RPC + enhanced-transactions APIs.
+// rpcURL is used for standard Solana JSON-RPC calls (getSignaturesForAddress) and can be
+// any Solana endpoint — we default to the public mainnet RPC to avoid consuming Helius credits
+// on raw pagination. enhURL is Helius-specific and costs credits; it is only called for the
+// 100-sig quality sample per poll.
 type Client struct {
 	httpClient *http.Client
 	apiKey     string
-	rpcURL     string // https://mainnet.helius-rpc.com/?api-key=KEY
-	enhURL     string // https://api.helius.xyz/v0/transactions?api-key=KEY
+	rpcURL     string // standard Solana JSON-RPC (public endpoint, no credits)
+	enhURL     string // https://api.helius.xyz/v0/transactions?api-key=KEY (credits)
 }
 
-func New(apiKey string) *Client {
+// NewWithRPC creates a client where rpcURL is used for standard RPC calls (getSignaturesForAddress).
+// Pass a public or self-hosted endpoint to avoid consuming Helius credits on pagination.
+func NewWithRPC(apiKey, rpcURL string) *Client {
 	return &Client{
 		httpClient: &http.Client{Timeout: 30 * time.Second},
 		apiKey:     apiKey,
-		rpcURL:     fmt.Sprintf("https://mainnet.helius-rpc.com/?api-key=%s", apiKey),
+		rpcURL:     rpcURL,
 		enhURL:     fmt.Sprintf("https://api.helius.xyz/v0/transactions?api-key=%s", apiKey),
 	}
 }
 
+func New(apiKey string) *Client {
+	return NewWithRPC(apiKey, "https://api.mainnet-beta.solana.com")
+}
+
 // SigEntry is one result from getSignaturesForAddress.
 type SigEntry struct {
 	Signature string `json:"signature"`
@@ -57,36 +67,57 @@ func (c *Client) GetSignaturesForAddress(ctx context.Context, address string, li
 		"params":  []any{address, params},
 	})
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.rpcURL, bytes.NewReader(body))
-	if err != nil {
-		return nil, err
-	}
-	req.Header.Set("Content-Type", "application/json")
-
-	resp, err := c.httpClient.Do(req)
-	if err != nil {
-		return nil, err
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("helius RPC HTTP %d", resp.StatusCode)
-	}
-
-	var out struct {
-		Result []SigEntry `json:"result"`
-		Error  *struct {
-			Code    int    `json:"code"`
-			Message string `json:"message"`
-		} `json:"error"`
-	}
-	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
-		return nil, fmt.Errorf("helius RPC decode: %w", err)
-	}
-	if out.Error != nil {
-		return nil, fmt.Errorf("helius RPC error %d: %s", out.Error.Code, out.Error.Message)
-	}
-	return out.Result, nil
+	// Retry up to 3 times on 429 or transient errors with exponential backoff.
+	var lastErr error
+	for attempt := range 3 {
+		if attempt > 0 {
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			case <-time.After(time.Duration(attempt*attempt) * time.Second):
+			}
+		}
+
+		req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.rpcURL, bytes.NewReader(body))
+		if err != nil {
+			return nil, err
+		}
+		req.Header.Set("Content-Type", "application/json")
+
+		resp, err := c.httpClient.Do(req)
+		if err != nil {
+			lastErr = err
+			continue
+		}
+
+		if resp.StatusCode == http.StatusTooManyRequests {
+			resp.Body.Close()
+			lastErr = fmt.Errorf("helius RPC HTTP 429")
+			continue
+		}
+		if resp.StatusCode != http.StatusOK {
+			resp.Body.Close()
+			return nil, fmt.Errorf("helius RPC HTTP %d", resp.StatusCode)
+		}
+
+		var out struct {
+			Result []SigEntry `json:"result"`
+			Error  *struct {
+				Code    int    `json:"code"`
+				Message string `json:"message"`
+			} `json:"error"`
+		}
+		if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
+			resp.Body.Close()
+			return nil, fmt.Errorf("helius RPC decode: %w", err)
+		}
+		resp.Body.Close()
+		if out.Error != nil {
+			return nil, fmt.Errorf("helius RPC error %d: %s", out.Error.Code, out.Error.Message)
+		}
+		return out.Result, nil
+	}
+	return nil, fmt.Errorf("helius RPC: %w (after 3 attempts)", lastErr)
 }
 
 // NativeTransfer is a SOL transfer extracted by Helius.

harnesses/solana-exec/migrations/003_cu_samples.sql

@@ -0,0 +1,16 @@
+BEGIN;
+
+-- Raw CU price samples from the 100-sig enhanced API poll each cycle.
+-- Gives real percentile_cont on actual data points instead of AVG(hourly_p50).
+-- Retention: 35 days (purged by materializer).
+CREATE TABLE IF NOT EXISTS solana_exec_cu_samples (
+    id             BIGSERIAL PRIMARY KEY,
+    platform       TEXT NOT NULL,
+    block_time     TIMESTAMPTZ NOT NULL,
+    cu_price_micro BIGINT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS solana_exec_cu_samples_platform_time
+    ON solana_exec_cu_samples (platform, block_time);
+
+COMMIT;

harnesses/solana-exec/migrations/004_raw_counts_idempotent.sql

@@ -0,0 +1,17 @@
+BEGIN;
+
+-- Make raw_counts idempotent on crash/restart.
+-- Problem: UpsertRawCounts accumulated (success_count + EXCLUDED.success_count).
+-- On a crash between UpsertRawCounts and SaveCursor, the next boot re-processes
+-- the same sigs and double-counts. Fix: key by (platform, bucket_start, from_cursor)
+-- so each poll's contribution is a distinct row. ON CONFLICT DO NOTHING = fully idempotent.
+
+ALTER TABLE solana_exec_raw_counts DROP CONSTRAINT solana_exec_raw_counts_pkey;
+
+-- from_cursor = cursor.LastSig at the start of the poll that produced this row.
+-- Existing rows get '' which is safe — they predate this schema change and won't be re-inserted.
+ALTER TABLE solana_exec_raw_counts ADD COLUMN IF NOT EXISTS from_cursor TEXT NOT NULL DEFAULT '';
+
+ALTER TABLE solana_exec_raw_counts ADD PRIMARY KEY (platform, bucket_start, from_cursor);
+
+COMMIT;

harnesses/solana-exec/migrations/005_events_purge_and_cu_sig.sql

@@ -0,0 +1,18 @@
+BEGIN;
+
+-- Bound events table growth: materializer will purge rows older than 90 days.
+-- Facts table already holds all aggregated metrics; raw events are only needed
+-- for re-materialization within the rolling window.
+-- No schema change needed — purge is done in application code.
+
+-- Make cu_samples idempotent on crash/restart by keying on (platform, sig).
+-- Without this, a crash between UpsertEvents and SaveCursor causes the next
+-- boot to re-call InsertCUSamples for the same 100 sigs, duplicating them
+-- and slightly skewing percentiles.
+ALTER TABLE solana_exec_cu_samples ADD COLUMN IF NOT EXISTS sig TEXT;
+
+CREATE UNIQUE INDEX IF NOT EXISTS solana_exec_cu_samples_platform_sig
+    ON solana_exec_cu_samples (platform, sig)
+    WHERE sig IS NOT NULL;
+
+COMMIT;

src/app/api/internal/codex-token/route.ts

@@ -0,0 +1,68 @@
+export const runtime = 'edge';
+
+const SECRET = process.env.INTERNAL_RELAY_SECRET ?? '';
+
+export async function GET(request: Request) {
+  const auth = request.headers.get('x-relay-secret');
+  if (SECRET && auth !== SECRET) {
+    return new Response('Forbidden', { status: 403 });
+  }
+
+  const body = JSON.stringify({
+    operationName: 'CreateApiToken',
+    query: 'mutation CreateApiToken { createApiTokens(input: { count: 1 }) { token } }',
+    variables: {},
+  });
+
+  const resp = await fetch('https://un.defined.fi/api', {
+    method: 'POST',
+    headers: {
+      'Content-Type': 'application/json',
+      Accept: 'application/json',
+      Origin: 'https://un.defined.fi',
+      Referer: 'https://un.defined.fi/',
+      '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',
+      'Accept-Language': 'en-US,en;q=0.9',
+      'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="131", "Google Chrome";v="131"',
+      'sec-ch-ua-mobile': '?0',
+      'sec-ch-ua-platform': '"macOS"',
+      'sec-fetch-dest': 'empty',
+      'sec-fetch-mode': 'cors',
+      'sec-fetch-site': 'same-origin',
+    },
+    body,
+  });
+
+  const status = resp.status;
+  const text = await resp.text();
+
+  if (status !== 200) {
+    return new Response(JSON.stringify({ error: `upstream ${status}`, body: text.slice(0, 200) }), {
+      status: 502,
+      headers: { 'Content-Type': 'application/json' },
+    });
+  }
+
+  let parsed: { data?: { createApiTokens?: { token: string }[] } };
+  try {
+    parsed = JSON.parse(text);
+  } catch {
+    return new Response(JSON.stringify({ error: 'parse error', body: text.slice(0, 200) }), {
+      status: 502,
+      headers: { 'Content-Type': 'application/json' },
+    });
+  }
+
+  const token = parsed?.data?.createApiTokens?.[0]?.token;
+  if (!token) {
+    return new Response(JSON.stringify({ error: 'no token', body: text.slice(0, 300) }), {
+      status: 502,
+      headers: { 'Content-Type': 'application/json' },
+    });
+  }
+
+  return new Response(token, {
+    headers: { 'Content-Type': 'text/plain' },
+  });
+}