web: add DZDP concentration view to geolocation explorer

CHANGELOG.md

@@ -0,0 +1,11 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
+
+## [Unreleased]
+
+### Added
+- DZDP concentration view in geolocation explorer with hero stats, anchor point map, country bar chart, ASN concentration list, and CTA banner (#550)
+- View switcher (QA Explorer / DZDP Concentration / DZDP Validators) with URL query param sync (#550)

api/handlers/geo_concentration.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log/slog"
 	"math"
 	"net/http"
 	"sort"
@@ -79,11 +80,13 @@ func (a *API) GetGeoConcentration(w http.ResponseWriter, r *http.Request) {
 }
 
 func (a *API) FetchGeoConcentrationData(ctx context.Context) (*GeoConcentrationResponse, error) {
-	ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
+	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
 
 	dzdpDB := fmt.Sprintf("`%s`", a.DZDPDB)
 
+	// Fetch enriched validators with lat/lng — nearest-metro assignment is done in Go
+	// to avoid a CROSS JOIN + geoDistance + arraySort in ClickHouse.
 	query := fmt.Sprintf(`
 		WITH geolocated AS (
 			SELECT
@@ -94,65 +97,32 @@ func (a *API) FetchGeoConcentrationData(ctx context.Context) (*GeoConcentrationR
 			FROM (SELECT * FROM %s.location_state FINAL) AS ls
 			JOIN solana_gossip_nodes_current gn ON ls.target_ip = gn.gossip_ip
 			WHERE ls.state = 'honed'
-		),
-		enriched AS (
-			SELECT
-				gv.target_ip AS target_ip,
-				gv.dzdp_lat AS dzdp_lat,
-				gv.dzdp_lng AS dzdp_lng,
-				va.vote_pubkey AS vote_pubkey,
-				va.activated_stake_lamports / 1e9 AS stake_sol,
-				coalesce(geo.asn, 0) AS asn,
-				coalesce(geo.asn_org, '') AS asn_org,
-				coalesce(geo.country_code, '') AS country_code,
-				coalesce(geo.country, '') AS country_name
-			FROM geolocated gv
-			JOIN solana_vote_accounts_current va ON gv.node_pubkey = va.node_pubkey
-			LEFT JOIN geoip_records_current geo ON gv.target_ip = geo.ip
-			WHERE va.epoch_vote_account = 'true' AND va.activated_stake_lamports > 0
-		),
-		nearest_metro AS (
-			SELECT
-				e.vote_pubkey AS vote_pubkey,
-				e.stake_sol AS stake_sol,
-				e.asn AS asn,
-				e.asn_org AS asn_org,
-				e.country_code AS country_code,
-				e.country_name AS country_name,
-				arrayElement(
-					arraySort(
-						(x, y) -> y,
-						groupArray(m.code),
-						groupArray(geoDistance(e.dzdp_lng, e.dzdp_lat, m.longitude, m.latitude))
-					), 1
-				) AS metro_code
-			FROM enriched e
-			CROSS JOIN dz_metros_current m
-			GROUP BY vote_pubkey, stake_sol, asn, asn_org, country_code, country_name
-		),
-		deduped AS (
-			SELECT
-				vote_pubkey,
-				max(stake_sol) AS max_stake,
-				argMax(metro_code, stake_sol) AS metro_code,
-				argMax(asn, stake_sol) AS asn,
-				argMax(asn_org, stake_sol) AS asn_org,
-				argMax(country_code, stake_sol) AS country_code,
-				argMax(country_name, stake_sol) AS country_name
-			FROM nearest_metro
-			GROUP BY vote_pubkey
 		)
-		SELECT vote_pubkey, max_stake AS stake_sol, metro_code, asn, asn_org, country_code, country_name
-		FROM deduped
+		SELECT
+			va.vote_pubkey,
+			va.activated_stake_lamports / 1e9 AS stake_sol,
+			gv.dzdp_lat,
+			gv.dzdp_lng,
+			coalesce(geo.asn, 0) AS asn,
+			coalesce(geo.asn_org, '') AS asn_org,
+			coalesce(geo.country_code, '') AS country_code,
+			coalesce(geo.country, '') AS country_name
+		FROM geolocated gv
+		JOIN solana_vote_accounts_current va ON gv.node_pubkey = va.node_pubkey
+		LEFT JOIN geoip_records_current geo ON gv.target_ip = geo.ip
+		WHERE va.epoch_vote_account = 'true' AND va.activated_stake_lamports > 0
 	`, dzdpDB)
 
 	start := time.Now()
 	rows, err := a.DB.Query(ctx, query)
 	metrics.RecordClickHouseQuery("geo_concentration", time.Since(start), err)
 	if err != nil {
-		// Return empty response when DZDP tables aren't available or accessible
+		// Return empty response when DZDP tables aren't available or accessible.
+		// Code 60 = UNKNOWN_TABLE, Code 81 = UNKNOWN_DATABASE, Code 497 = NOT_ENOUGH_PRIVILEGES.
 		var chErr *proto.Exception
-		if errors.As(err, &chErr) && (chErr.Code == 60 || chErr.Code == 497) {
+		if errors.As(err, &chErr) && (chErr.Code == 60 || chErr.Code == 81 || chErr.Code == 497) {
+			slog.Warn("geo concentration: DZDP tables not available, returning empty response",
+				"dzdp_db", dzdpDB, "ch_error_code", chErr.Code, "error", err)
 			return &GeoConcentrationResponse{
 				Metros:    []GeoConcentrationMetro{},
 				Countries: []GeoConcentrationCountry{},
@@ -162,32 +132,94 @@ func (a *API) FetchGeoConcentrationData(ctx context.Context) (*GeoConcentrationR
 		return nil, err
 	}
 
-	// Collect per-validator rows and aggregate in Go
-	type validatorRow struct {
+	type enrichedRow struct {
 		votePubkey  string
 		stakeSol    float64
-		metroCode   string
+		lat, lng    float64
 		asn         int64
 		asnOrg      string
 		countryCode string
 		countryName string
 	}
 
-	var validators []validatorRow
+	var enriched []enrichedRow
 	for rows.Next() {
-		var v validatorRow
-		if err := rows.Scan(&v.votePubkey, &v.stakeSol, &v.metroCode, &v.asn, &v.asnOrg, &v.countryCode, &v.countryName); err != nil {
+		var r enrichedRow
+		if err := rows.Scan(&r.votePubkey, &r.stakeSol, &r.lat, &r.lng, &r.asn, &r.asnOrg, &r.countryCode, &r.countryName); err != nil {
 			rows.Close()
 			return nil, err
 		}
-		validators = append(validators, v)
+		enriched = append(enriched, r)
 	}
 	if err := rows.Err(); err != nil {
 		rows.Close()
 		return nil, err
 	}
 	rows.Close()
 
+	// Fetch metros for nearest-metro assignment and anchor point count.
+	type metro struct {
+		code     string
+		lat, lng float64
+	}
+	var metros_list []metro
+	metroRows, err := a.DB.Query(ctx, "SELECT code, latitude, longitude FROM dz_metros_current")
+	if err != nil {
+		logError("geo concentration metros query error", "error", err)
+	} else {
+		for metroRows.Next() {
+			var m metro
+			if err := metroRows.Scan(&m.code, &m.lat, &m.lng); err != nil {
+				metroRows.Close()
+				return nil, err
+			}
+			metros_list = append(metros_list, m)
+		}
+		metroRows.Close()
+	}
+
+	// Assign nearest metro and deduplicate by vote_pubkey in Go.
+	type validatorRow struct {
+		votePubkey  string
+		stakeSol    float64
+		metroCode   string
+		asn         int64
+		asnOrg      string
+		countryCode string
+		countryName string
+	}
+
+	deduped := make(map[string]validatorRow)
+	for _, e := range enriched {
+		// Find nearest metro using Haversine distance.
+		bestCode := ""
+		bestDist := math.MaxFloat64
+		for _, m := range metros_list {
+			d := haversine(e.lat, e.lng, m.lat, m.lng)
+			if d < bestDist {
+				bestDist = d
+				bestCode = m.code
+			}
+		}
+
+		if prev, ok := deduped[e.votePubkey]; !ok || e.stakeSol > prev.stakeSol {
+			deduped[e.votePubkey] = validatorRow{
+				votePubkey:  e.votePubkey,
+				stakeSol:    e.stakeSol,
+				metroCode:   bestCode,
+				asn:         e.asn,
+				asnOrg:      e.asnOrg,
+				countryCode: e.countryCode,
+				countryName: e.countryName,
+			}
+		}
+	}
+
+	validators := make([]validatorRow, 0, len(deduped))
+	for _, v := range deduped {
+		validators = append(validators, v)
+	}
+
 	// Compute total stake
 	var totalStake float64
 	for _, v := range validators {
@@ -267,18 +299,6 @@ func (a *API) FetchGeoConcentrationData(ctx context.Context) (*GeoConcentrationR
 		stakeTopTwo += metros[i].StakePct
 	}
 
-	// Count anchor points (distinct DZ metros)
-	var anchorPoints uint64
-	anchorRows, err := a.DB.Query(ctx, "SELECT count() FROM dz_metros_current")
-	if err != nil {
-		logError("geo concentration anchor points query error", "error", err)
-	} else {
-		if anchorRows.Next() {
-			_ = anchorRows.Scan(&anchorPoints)
-		}
-		anchorRows.Close()
-	}
-
 	var maxASNPct float64
 	if len(asns) > 0 {
 		maxASNPct = asns[0].StakePct
@@ -302,7 +322,7 @@ func (a *API) FetchGeoConcentrationData(ctx context.Context) (*GeoConcentrationR
 		HeroStats: GeoConcentrationHeroStats{
 			ValidatorsMeasured:   len(validators),
 			StakeTopTwoMetrosPct: stakeTopTwo,
-			AnchorPoints:         int(anchorPoints),
+			AnchorPoints:         len(metros_list),
 			StakeMaxASNPct:       maxASNPct,
 		},
 		Metros:    metros,
@@ -312,3 +332,14 @@ func (a *API) FetchGeoConcentrationData(ctx context.Context) (*GeoConcentrationR
 
 	return resp, nil
 }
+
+// haversine returns the great-circle distance in meters between two points.
+func haversine(lat1, lng1, lat2, lng2 float64) float64 {
+	const earthRadius = 6_371_000 // meters
+	dLat := (lat2 - lat1) * math.Pi / 180
+	dLng := (lng2 - lng1) * math.Pi / 180
+	lat1r := lat1 * math.Pi / 180
+	lat2r := lat2 * math.Pi / 180
+	a := math.Sin(dLat/2)*math.Sin(dLat/2) + math.Cos(lat1r)*math.Cos(lat2r)*math.Sin(dLng/2)*math.Sin(dLng/2)
+	return earthRadius * 2 * math.Atan2(math.Sqrt(a), math.Sqrt(1-a))
+}