ai/mcp: geoip

Author: snormore

telemetry/global-monitor/internal/gm/planner_dz_icmp.go

@@ -223,7 +223,7 @@ func (p *DoubleZeroUserICMPPlanner) recordResult(source *Source, user *dz.User,
 		tags["target_geoip_region"] = geoIP.Region
 		tags["target_geoip_city"] = geoIP.City
 		tags["target_geoip_city_id"] = strconv.Itoa(geoIP.CityID)
-		tags["target_geoip_metro"] = geoIP.Metro
+		tags["target_geoip_metro"] = geoIP.MetroName
 		tags["target_geoip_asn"] = strconv.Itoa(int(geoIP.ASN))
 		tags["target_geoip_asn_org"] = geoIP.ASNOrg
 		fields["target_geoip_latitude"] = geoIP.Latitude

telemetry/global-monitor/internal/gm/planner_dz_icmp_test.go

@@ -166,7 +166,7 @@ func TestGlobalMonitor_DoubleZeroUserICMPPlanner_Record_WritesExpectedInfluxPoin
 			Region:      "ON",
 			City:        "Toronto",
 			CityID:      123,
-			Metro:       "YYZ",
+			MetroName:   "Yorkton",
 			ASN:         64500,
 			ASNOrg:      "Example",
 			Latitude:    43.7,

telemetry/global-monitor/internal/gm/planner_sol_icmp.go

@@ -228,7 +228,7 @@ func (p *SolanaValidatorICMPPlanner) recordResult(source *Source, val *sol.Valid
 		tags["target_geoip_region"] = val.GeoIP.Region
 		tags["target_geoip_city"] = val.GeoIP.City
 		tags["target_geoip_city_id"] = strconv.Itoa(val.GeoIP.CityID)
-		tags["target_geoip_metro"] = val.GeoIP.Metro
+		tags["target_geoip_metro"] = val.GeoIP.MetroName
 		tags["target_geoip_asn"] = strconv.Itoa(int(val.GeoIP.ASN))
 		tags["target_geoip_asn_org"] = val.GeoIP.ASNOrg
 		fields["target_geoip_latitude"] = val.GeoIP.Latitude

telemetry/global-monitor/internal/gm/planner_sol_icmp_test.go

@@ -165,7 +165,7 @@ func TestGlobalMonitor_SolanaValidatorICMPPlanner_Record_WritesExpectedInfluxPoi
 		Region:      "ON",
 		City:        "Toronto",
 		CityID:      123,
-		Metro:       "YYZ",
+		MetroName:   "Yorkton",
 		ASN:         64500,
 		ASNOrg:      "Example",
 		Latitude:    43.7,

telemetry/global-monitor/internal/gm/planner_sol_tpuquic.go

@@ -251,7 +251,7 @@ func (p *SolanaValidatorTPUQUICPlanner) recordResult(source *Source, val *sol.Va
 		tags["target_geoip_region"] = val.GeoIP.Region
 		tags["target_geoip_city"] = val.GeoIP.City
 		tags["target_geoip_city_id"] = strconv.Itoa(val.GeoIP.CityID)
-		tags["target_geoip_metro"] = val.GeoIP.Metro
+		tags["target_geoip_metro"] = val.GeoIP.MetroName
 		tags["target_geoip_asn"] = strconv.Itoa(int(val.GeoIP.ASN))
 		tags["target_geoip_asn_org"] = val.GeoIP.ASNOrg
 		fields["target_geoip_latitude"] = val.GeoIP.Latitude

telemetry/global-monitor/internal/gm/planner_sol_tpuquic_test.go

@@ -162,7 +162,7 @@ func TestGlobalMonitor_SolanaValidatorTPUQUICPlanner_Record_WritesExpectedInflux
 		Region:      "ON",
 		City:        "Toronto",
 		CityID:      123,
-		Metro:       "YYZ",
+		MetroName:   "Yorkton",
 		ASN:         64500,
 		ASNOrg:      "Example",
 		Latitude:    43.7,

tools/dz-ai/cmd/mcp-server/main.go

@@ -29,7 +29,10 @@ import (
 	"github.com/malbeclabs/doublezero/tools/dz-ai/internal/mcp/duck"
 	"github.com/malbeclabs/doublezero/tools/dz-ai/internal/mcp/metrics"
 	"github.com/malbeclabs/doublezero/tools/dz-ai/internal/mcp/server"
+	"github.com/malbeclabs/doublezero/tools/maxmind/pkg/geoip"
+	"github.com/malbeclabs/doublezero/tools/maxmind/pkg/metrodb"
 	"github.com/malbeclabs/doublezero/tools/solana/pkg/rpc"
+	"github.com/oschwald/geoip2-golang"
 )
 
 var (
@@ -45,10 +48,13 @@ const (
 	defaultMaxConcurrency  = 64
 	defaultDZEnv           = config.EnvMainnetBeta
 	defaultMetricsAddr     = "0.0.0.0:8080"
-	// defaultDBPath          = ".tmp/mcp.duckdb"
 	defaultDBPath          = ":memory:"
 	defaultDuckDBSpillPath = ".tmp/duckdb-spill-tmp"
 	defaultDBPathEnvVar    = "MCP_DB_PATH"
+	defaultGeoipCityDBPath = "/usr/share/GeoIP/GeoLite2-City.mmdb"
+	defaultGeoipASNDBPath  = "/usr/share/GeoIP/GeoLite2-ASN.mmdb"
+	geoipCityDBPathEnvVar  = "MCP_GEOIP_CITY_DB_PATH"
+	geoipASNDBPathEnvVar   = "MCP_GEOIP_ASN_DB_PATH"
 )
 
 func main() {
@@ -69,6 +75,8 @@ func run() error {
 	metricsAddrFlag := flag.String("metrics-addr", defaultMetricsAddr, "Address to listen on for prometheus metrics")
 	dbPathFlag := flag.String("db-path", defaultDBPath, "Path to DuckDB database file (empty for in-memory, or set MCP_DB_PATH env var)")
 	dbSpillDirFlag := flag.String("db-spill-dir", defaultDuckDBSpillPath, "Path to DuckDB temporary spill directory")
+	geoipCityDBPathFlag := flag.String("geoip-city-db-path", defaultGeoipCityDBPath, "Path to MaxMind GeoIP2 City database file (or set MCP_GEOIP_CITY_DB_PATH env var)")
+	geoipASNDBPathFlag := flag.String("geoip-asn-db-path", defaultGeoipASNDBPath, "Path to MaxMind GeoIP2 ASN database file (or set MCP_GEOIP_ASN_DB_PATH env var)")
 	flag.Parse()
 
 	networkConfig, err := config.NetworkConfigForEnv(*envFlag)
@@ -137,8 +145,10 @@ func run() error {
 
 	// Determine database path: flag takes precedence, then env var, then default
 	dbPath := *dbPathFlag
-	if dbPath == "" {
-		dbPath = os.Getenv(defaultDBPathEnvVar)
+	if dbPath == "" || dbPath == defaultDBPath {
+		if envPath := os.Getenv(defaultDBPathEnvVar); envPath != "" {
+			dbPath = envPath
+		}
 	}
 	if dbPath == "" {
 		dbPath = defaultDBPath
@@ -151,6 +161,28 @@ func run() error {
 	}
 	defer dbCloseFn()
 
+	// Determine GeoIP database paths: flag takes precedence, then env var, then default
+	geoipCityDBPath := *geoipCityDBPathFlag
+	if geoipCityDBPath == defaultGeoipCityDBPath {
+		if envPath := os.Getenv(geoipCityDBPathEnvVar); envPath != "" {
+			geoipCityDBPath = envPath
+		}
+	}
+
+	geoipASNDBPath := *geoipASNDBPathFlag
+	if geoipASNDBPath == defaultGeoipASNDBPath {
+		if envPath := os.Getenv(geoipASNDBPathEnvVar); envPath != "" {
+			geoipASNDBPath = envPath
+		}
+	}
+
+	// Initialize GeoIP resolver
+	geoIPResolver, geoIPCloseFn, err := initializeGeoIP(geoipCityDBPath, geoipASNDBPath, log)
+	if err != nil {
+		return fmt.Errorf("failed to initialize GeoIP: %w", err)
+	}
+	defer geoIPCloseFn()
+
 	// Parse allowed tokens from environment variable (comma-separated)
 	// Auth can be explicitly disabled with MCP_AUTH_DISABLED=true
 	var allowedTokens []string
@@ -188,6 +220,7 @@ func run() error {
 		InternetLatencyAgentPK: networkConfig.InternetLatencyCollectorPK,
 		InternetDataProviders:  telemetryconfig.InternetTelemetryDataProviders,
 		AllowedTokens:          allowedTokens,
+		GeoIPResolver:          geoIPResolver,
 	})
 	if err != nil {
 		return fmt.Errorf("failed to create server: %w", err)
@@ -273,3 +306,40 @@ func initializeDuckDB(dbPath string, spillDir string, log *slog.Logger) (duck.DB
 		return nil
 	}, nil
 }
+
+func initializeGeoIP(cityDBPath, asnDBPath string, log *slog.Logger) (geoip.Resolver, func() error, error) {
+	cityDB, err := geoip2.Open(cityDBPath)
+	if err != nil {
+		return nil, nil, fmt.Errorf("failed to open GeoIP city database: %w", err)
+	}
+
+	asnDB, err := geoip2.Open(asnDBPath)
+	if err != nil {
+		cityDB.Close()
+		return nil, nil, fmt.Errorf("failed to open GeoIP ASN database: %w", err)
+	}
+
+	metroDB, err := metrodb.New()
+	if err != nil {
+		cityDB.Close()
+		asnDB.Close()
+		return nil, nil, fmt.Errorf("failed to create metro database: %w", err)
+	}
+
+	resolver, err := geoip.NewResolver(log, cityDB, asnDB, metroDB)
+	if err != nil {
+		cityDB.Close()
+		asnDB.Close()
+		return nil, nil, fmt.Errorf("failed to create GeoIP resolver: %w", err)
+	}
+
+	return resolver, func() error {
+		if err := cityDB.Close(); err != nil {
+			return fmt.Errorf("failed to close city database: %w", err)
+		}
+		if err := asnDB.Close(); err != nil {
+			return fmt.Errorf("failed to close ASN database: %w", err)
+		}
+		return nil
+	}, nil
+}

tools/dz-ai/internal/agent/system_prompt.go

@@ -26,6 +26,7 @@ ANSWERING RULES:
 - Latency units: display in milliseconds (ms) by default; use microseconds (µs) only when values are < 0.1 ms.
 - Drain semantics: treat dz_links.delay_override_ns = 1000000000 as soft-drained when interpreting link state.
 - Link health: consider drained, telemetry packet loss, and delay delta from committed delay when interpreting link health.
+- User location: use geoip data and connected devices but tell the user that's how it was determined.
 
 OUTPUT STYLE (MANDATORY):
 - Always structure responses using section headers, even for short answers.

tools/dz-ai/internal/mcp/duck/upsert.go

@@ -0,0 +1,121 @@
+package duck
+
+import (
+	"context"
+	"database/sql"
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"time"
+)
+
+func UpsertTableViaCSV(ctx context.Context, log *slog.Logger, db DB, tableName string, count int, writeCSVFn func(*csv.Writer, int) error) error {
+	upsertStart := time.Now()
+	defer func() {
+		duration := time.Since(upsertStart)
+		log.Debug("upserting to table completed", "table", tableName, "rows", count, "duration", duration.String())
+	}()
+
+	if count == 0 {
+		return nil
+	}
+
+	// Create a temporary CSV file for COPY FROM (much faster than individual INSERTs)
+	tmpFile, err := os.CreateTemp("", fmt.Sprintf("%s_upsert_*.csv", tableName))
+	if err != nil {
+		return fmt.Errorf("failed to create temp file: %w", err)
+	}
+	defer os.Remove(tmpFile.Name())
+	defer tmpFile.Close()
+
+	// Write CSV data
+	csvWriter := csv.NewWriter(tmpFile)
+	csvWriter.Comma = ','
+
+	// Log progress every 5 seconds for long-running operations
+	progressLogInterval := 5 * time.Second
+	lastProgressLog := time.Now()
+
+	for i := range count {
+		// Check for context cancellation during long-running write operations
+		select {
+		case <-ctx.Done():
+			return fmt.Errorf("context cancelled while writing CSV for %s: %w", tableName, ctx.Err())
+		default:
+		}
+
+		if err := writeCSVFn(csvWriter, i); err != nil {
+			log.Error("failed to write CSV record", "table", tableName, "row", i, "total", count, "error", err)
+			return fmt.Errorf("failed to write CSV record for %s: %w", tableName, err)
+		}
+
+		// Log progress periodically for large operations
+		if count > 1000 {
+			now := time.Now()
+			if now.Sub(lastProgressLog) >= progressLogInterval {
+				log.Debug("write progress", "table", tableName, "written", i+1, "total", count)
+				lastProgressLog = now
+			}
+		}
+	}
+
+	csvWriter.Flush()
+	if err := csvWriter.Error(); err != nil {
+		return fmt.Errorf("CSV writer error: %w", err)
+	}
+	if err := tmpFile.Sync(); err != nil {
+		return fmt.Errorf("failed to sync temp file: %w", err)
+	}
+
+	// Close file before COPY (DuckDB needs to open it)
+	tmpFile.Close()
+
+	// Check for context cancellation before starting transaction
+	select {
+	case <-ctx.Done():
+		return fmt.Errorf("context cancelled before transaction for %s: %w", tableName, ctx.Err())
+	default:
+	}
+
+	tx, err := db.Begin()
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction for %s: %w", tableName, err)
+	}
+	defer func() {
+		if err := tx.Rollback(); err != nil && !errors.Is(err, sql.ErrTxDone) {
+			log.Error("failed to rollback transaction", "table", tableName, "error", err)
+		}
+	}()
+
+	// Create temporary table with same structure
+	// Use DROP IF EXISTS to handle case where temp table already exists from previous failed transaction
+	tempTableName := fmt.Sprintf("%s_temp_upsert", tableName)
+	dropTempSQL := fmt.Sprintf(`DROP TABLE IF EXISTS %s`, tempTableName)
+	if _, err := tx.Exec(dropTempSQL); err != nil {
+		return fmt.Errorf("failed to drop temp table for %s: %w", tableName, err)
+	}
+	createTempSQL := fmt.Sprintf(`CREATE TEMP TABLE %s AS SELECT * FROM %s WHERE 1=0`, tempTableName, tableName)
+	if _, err := tx.Exec(createTempSQL); err != nil {
+		return fmt.Errorf("failed to create temp table for %s: %w", tableName, err)
+	}
+
+	// Load CSV into temporary table
+	copySQL := fmt.Sprintf("COPY %s FROM '%s' (FORMAT CSV, HEADER false)", tempTableName, tmpFile.Name())
+	if _, err := tx.Exec(copySQL); err != nil {
+		return fmt.Errorf("failed to COPY FROM CSV for %s: %w", tableName, err)
+	}
+
+	// Use INSERT OR REPLACE to merge temp table into main table
+	upsertSQL := fmt.Sprintf(`INSERT OR REPLACE INTO %s SELECT * FROM %s`, tableName, tempTableName)
+	if _, err := tx.Exec(upsertSQL); err != nil {
+		return fmt.Errorf("failed to upsert from temp table for %s: %w", tableName, err)
+	}
+
+	if err := tx.Commit(); err != nil {
+		log.Error("transaction commit failed", "table", tableName, "error", err)
+		return fmt.Errorf("failed to commit transaction for %s: %w", tableName, err)
+	}
+	return nil
+}