better pds scan & endpoint validation

cmd/import-labels/main.go

···
 		if lineCount%100000 == 0 {
 			elapsed := time.Since(startTime).Seconds()
 			rate := float64(lineCount) / elapsed
-			fmt.Printf("  ... processed %,d lines (%.0f lines/sec)\n", lineCount, rate)
+			fmt.Printf("  ... processed %d lines (%.0f lines/sec)\n", lineCount, rate)
 		}
 	}
 
···
 	fmt.Println("Import Summary")
 	fmt.Println("========================================")
 	fmt.Printf("✓ Import completed in %v\n", totalTime)
-	fmt.Printf("Total lines processed: %,d\n", lineCount)
+	fmt.Printf("Total lines processed: %d\n", lineCount)
 }

go.mod

···
 require github.com/klauspost/compress v1.18.1
 
 require (
-	github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d
 	github.com/gorilla/handlers v1.5.2
 	github.com/jackc/pgx/v5 v5.7.6
 	tangled.org/atscan.net/plcbundle v0.3.6

go.sum

···
-github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8=
-github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

internal/api/handlers.go

···
 		"endpoint":      pds.Endpoint,
 		"discovered_at": pds.DiscoveredAt,
 		"status":        statusToString(pds.Status),
+		"valid":         pds.Valid, // NEW
 	}
 
 	// Add server_did if available

internal/pds/client.go

···
 }
 
 // DescribeServer fetches com.atproto.server.describeServer
-func (c *Client) DescribeServer(ctx context.Context, endpoint string) (*ServerDescription, error) {
+// Returns: description, responseTime, usedIP, error
+func (c *Client) DescribeServer(ctx context.Context, endpoint string) (*ServerDescription, time.Duration, string, error) {
+	startTime := time.Now()
 	url := fmt.Sprintf("%s/xrpc/com.atproto.server.describeServer", endpoint)
 
-	//fmt.Println(url)
+	// Track which IP was used
+	var usedIP string
+	transport := &http.Transport{
+		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
+			conn, err := (&net.Dialer{
+				Timeout:   30 * time.Second,
+				KeepAlive: 30 * time.Second,
+			}).DialContext(ctx, network, addr)
+
+			if err == nil && conn != nil {
+				if remoteAddr := conn.RemoteAddr(); remoteAddr != nil {
+					if tcpAddr, ok := remoteAddr.(*net.TCPAddr); ok {
+						usedIP = tcpAddr.IP.String()
+					}
+				}
+			}
+			return conn, err
+		},
+	}
+
+	client := &http.Client{
+		Timeout:   c.httpClient.Timeout,
+		Transport: transport,
+	}
 
 	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
 	if err != nil {
-		return nil, err
+		return nil, 0, "", err
 	}
 
-	resp, err := c.httpClient.Do(req)
+	resp, err := client.Do(req)
+	responseTime := time.Since(startTime)
+
 	if err != nil {
-		return nil, err
+		return nil, responseTime, usedIP, err
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		return nil, responseTime, usedIP, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
 	}
 
 	var desc ServerDescription
 	if err := json.NewDecoder(resp.Body).Decode(&desc); err != nil {
-		return nil, err
+		return nil, responseTime, usedIP, err
 	}
 
-	return &desc, nil
+	return &desc, responseTime, usedIP, nil
 }
 
 // CheckHealth performs a basic health check, ensuring the endpoint returns JSON with a "version"
-// Returns: available, responseTime, version, usedIP, error
-func (c *Client) CheckHealth(ctx context.Context, endpoint string) (bool, time.Duration, string, string, error) {
+// Returns: available, responseTime, version, error
+func (c *Client) CheckHealth(ctx context.Context, endpoint string) (bool, time.Duration, string, error) {
 	startTime := time.Now()
 
 	url := fmt.Sprintf("%s/xrpc/_health", endpoint)
 	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
 	if err != nil {
-		return false, 0, "", "", err
-	}
-
-	// Create a custom dialer to track which IP was actually used
-	var usedIP string
-	transport := &http.Transport{
-		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
-			conn, err := (&net.Dialer{
-				Timeout:   30 * time.Second,
-				KeepAlive: 30 * time.Second,
-			}).DialContext(ctx, network, addr)
-
-			if err == nil && conn != nil {
-				if remoteAddr := conn.RemoteAddr(); remoteAddr != nil {
-					// Extract IP from "ip:port" format
-					if tcpAddr, ok := remoteAddr.(*net.TCPAddr); ok {
-						usedIP = tcpAddr.IP.String()
-					}
-				}
-			}
-
-			return conn, err
-		},
+		return false, 0, "", err
 	}
 
-	// Create a client with our custom transport
-	client := &http.Client{
-		Timeout:   c.httpClient.Timeout,
-		Transport: transport,
-	}
-
-	resp, err := client.Do(req)
+	resp, err := c.httpClient.Do(req)
 	duration := time.Since(startTime)
 
 	if err != nil {
-		return false, duration, "", usedIP, err
+		return false, duration, "", err
 	}
 	defer resp.Body.Close()
 
 	if resp.StatusCode != http.StatusOK {
-		return false, duration, "", usedIP, fmt.Errorf("health check returned status %d", resp.StatusCode)
+		return false, duration, "", fmt.Errorf("health check returned status %d", resp.StatusCode)
 	}
 
 	// Decode the JSON response and check for "version"
···
 	}
 
 	if err := json.NewDecoder(resp.Body).Decode(&healthResponse); err != nil {
-		return false, duration, "", usedIP, fmt.Errorf("failed to decode health JSON: %w", err)
+		return false, duration, "", fmt.Errorf("failed to decode health JSON: %w", err)
 	}
 
 	if healthResponse.Version == "" {
-		return false, duration, "", usedIP, fmt.Errorf("health JSON response missing 'version' field")
+		return false, duration, "", fmt.Errorf("health JSON response missing 'version' field")
 	}
 
 	// All checks passed
-	return true, duration, healthResponse.Version, usedIP, nil
+	return true, duration, healthResponse.Version, nil
 }

internal/pds/scanner.go

···
 	"sync/atomic"
 	"time"
 
-	"github.com/acarl005/stripansi"
 	"github.com/atscan/atscand/internal/config"
 	"github.com/atscan/atscand/internal/ipinfo"
 	"github.com/atscan/atscand/internal/log"
···
 	servers, err := s.db.GetEndpoints(ctx, &storage.EndpointFilter{
 		Type:            "pds",
 		OnlyStale:       true,
+		OnlyValid:       true,
 		RecheckInterval: s.config.RecheckInterval,
 	})
 	if err != nil {
···
 	// STEP 1: Resolve IPs (both IPv4 and IPv6)
 	ips, err := ipinfo.ExtractIPsFromEndpoint(ep.Endpoint)
 	if err != nil {
-		// Mark as offline due to DNS failure
 		s.saveScanResult(ctx, ep.ID, &ScanResult{
 			Status:       storage.EndpointStatusOffline,
 			ErrorMessage: fmt.Sprintf("DNS resolution failed: %v", err),
···
 		go s.updateIPInfoIfNeeded(ctx, ips.IPv6)
 	}
 
-	// STEP 2: Health check (now returns which IP was used)
-	available, responseTime, version, usedIP, err := s.client.CheckHealth(ctx, ep.Endpoint)
-	if err != nil || !available {
-		errMsg := "health check failed"
-		if err != nil {
-			errMsg = err.Error()
-		}
+	// STEP 2: Call describeServer (primary health check + metadata)
+	desc, descResponseTime, usedIP, err := s.client.DescribeServer(ctx, ep.Endpoint)
+	if err != nil {
 		s.saveScanResult(ctx, ep.ID, &ScanResult{
 			Status:       storage.EndpointStatusOffline,
-			ResponseTime: responseTime,
-			ErrorMessage: errMsg,
-			UsedIP:       usedIP, // Save even if failed
+			ResponseTime: descResponseTime,
+			ErrorMessage: fmt.Sprintf("describeServer failed: %v", err),
+			UsedIP:       usedIP,
 		})
 		return
 	}
 
-	// STEP 3: Fetch PDS-specific data
-	desc, err := s.client.DescribeServer(ctx, ep.Endpoint)
-	if err != nil {
-		log.Verbose("Warning: failed to describe server %s: %v", stripansi.Strip(ep.Endpoint), err)
-	} else if desc != nil && desc.DID != "" {
+	// Update server DID immediately
+	if desc.DID != "" {
 		s.db.UpdateEndpointServerDID(ctx, ep.ID, desc.DID)
 	}
 
-	// Fetch repos with full info
+	// STEP 3: Call _health to get version
+	available, healthResponseTime, version, err := s.client.CheckHealth(ctx, ep.Endpoint)
+	if err != nil || !available {
+		log.Verbose("Warning: _health check failed for %s: %v", ep.Endpoint, err)
+		// Server is online (describeServer worked) but _health failed
+		// Continue with empty version
+		version = ""
+	}
+
+	// Calculate average response time from both calls
+	avgResponseTime := descResponseTime
+	if available {
+		avgResponseTime = (descResponseTime + healthResponseTime) / 2
+	}
+
+	// STEP 4: Fetch repos
 	repoList, err := s.client.ListRepos(ctx, ep.Endpoint)
 	if err != nil {
 		log.Verbose("Warning: failed to list repos for %s: %v", ep.Endpoint, err)
 		repoList = []Repo{}
 	}
 
-	// Convert to DIDs for backward compatibility
+	// Convert to DIDs
 	dids := make([]string, len(repoList))
 	for i, repo := range repoList {
 		dids[i] = repo.DID
 	}
 
-	// STEP 4: SAVE scan result
+	// STEP 5: SAVE scan result
 	s.saveScanResult(ctx, ep.ID, &ScanResult{
 		Status:       storage.EndpointStatusOnline,
-		ResponseTime: responseTime,
+		ResponseTime: avgResponseTime,
 		Description:  desc,
 		DIDs:         dids,
 		Version:      version,
-		UsedIP:       usedIP, // NEW: Save which IP was used
+		UsedIP:       usedIP, // Only from describeServer
 	})
 
-	// Save repos in batches (only tracks changes)
+	// STEP 6: Save repos in batches (only tracks changes)
 	if len(repoList) > 0 {
 		batchSize := 100_000
 
···
 
 		log.Verbose("✓ Processed %d repos for %s", len(repoList), ep.Endpoint)
 	}
-
-	// IP info fetch already started at the beginning (step 1.5)
-	// It will complete in the background
 }
 
 func (s *Scanner) saveScanResult(ctx context.Context, endpointID int64, result *ScanResult) {

internal/plc/scanner.go

···
 }
 
 func (s *Scanner) storeEndpoint(ctx context.Context, epType, endpoint string, discoveredAt time.Time) error {
+	valid := validateEndpoint(endpoint)
 	return s.db.UpsertEndpoint(ctx, &storage.Endpoint{
 		EndpointType: epType,
 		Endpoint:     endpoint,
 		DiscoveredAt: discoveredAt,
 		LastChecked:  time.Time{},
 		Status:       storage.EndpointStatusUnknown,
+		Valid:        valid,
 	})
 }
 

internal/plc/types.go

···
 package plc
 
 import (
+	"net/url"
+	"strings"
+
 	plclib "tangled.org/atscan.net/plcbundle/plc"
 )
 
···
 	Confidence float64  `json:"confidence"`
 	Detectors  []string `json:"detectors"`
 }
+
+// validateEndpoint checks if endpoint is in correct format: https://<domain>
+func validateEndpoint(endpoint string) bool {
+	// Must not be empty
+	if endpoint == "" {
+		return false
+	}
+
+	// Must not have trailing slash
+	if strings.HasSuffix(endpoint, "/") {
+		return false
+	}
+
+	// Parse URL
+	u, err := url.Parse(endpoint)
+	if err != nil {
+		return false
+	}
+
+	// Must use https scheme
+	if u.Scheme != "https" {
+		return false
+	}
+
+	// Must have a host
+	if u.Host == "" {
+		return false
+	}
+
+	// Must not have path (except empty)
+	if u.Path != "" && u.Path != "/" {
+		return false
+	}
+
+	// Must not have query parameters
+	if u.RawQuery != "" {
+		return false
+	}
+
+	// Must not have fragment
+	if u.Fragment != "" {
+		return false
+	}
+
+	return true
+}

internal/storage/postgres.go

···
         ip TEXT,
         ipv6 TEXT,
         ip_resolved_at TIMESTAMP,
+		valid BOOLEAN DEFAULT true,
         updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
         UNIQUE(endpoint_type, endpoint)
     );
···
     CREATE INDEX IF NOT EXISTS idx_endpoints_ipv6 ON endpoints(ipv6);
     CREATE INDEX IF NOT EXISTS idx_endpoints_server_did ON endpoints(server_did);
     CREATE INDEX IF NOT EXISTS idx_endpoints_server_did_type_discovered ON endpoints(server_did, endpoint_type, discovered_at);
+	CREATE INDEX IF NOT EXISTS idx_endpoints_valid ON endpoints(valid);
 
 	-- IP infos table (IP as PRIMARY KEY)
 	CREATE TABLE IF NOT EXISTS ip_infos (
···
 
 func (p *PostgresDB) UpsertEndpoint(ctx context.Context, endpoint *Endpoint) error {
 	query := `
-        INSERT INTO endpoints (endpoint_type, endpoint, discovered_at, last_checked, status, ip, ipv6, ip_resolved_at)
-        VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
+        INSERT INTO endpoints (endpoint_type, endpoint, discovered_at, last_checked, status, ip, ipv6, ip_resolved_at, valid)
+        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
         ON CONFLICT(endpoint_type, endpoint) DO UPDATE SET
             last_checked = EXCLUDED.last_checked,
             status = EXCLUDED.status,
···
                 WHEN (EXCLUDED.ip IS NOT NULL AND EXCLUDED.ip != '') OR (EXCLUDED.ipv6 IS NOT NULL AND EXCLUDED.ipv6 != '') THEN EXCLUDED.ip_resolved_at
                 ELSE endpoints.ip_resolved_at
             END,
+            valid = EXCLUDED.valid,
             updated_at = CURRENT_TIMESTAMP
         RETURNING id
     `
 	err := p.db.QueryRowContext(ctx, query,
 		endpoint.EndpointType, endpoint.Endpoint, endpoint.DiscoveredAt,
-		endpoint.LastChecked, endpoint.Status, endpoint.IP, endpoint.IPv6, endpoint.IPResolvedAt).Scan(&endpoint.ID)
+		endpoint.LastChecked, endpoint.Status, endpoint.IP, endpoint.IPv6, endpoint.IPResolvedAt, endpoint.Valid).Scan(&endpoint.ID)
 	return err
 }
 
···
 func (p *PostgresDB) GetEndpoint(ctx context.Context, endpoint string, endpointType string) (*Endpoint, error) {
 	query := `
         SELECT id, endpoint_type, endpoint, discovered_at, last_checked, status, 
-               ip, ipv6, ip_resolved_at, updated_at
+               ip, ipv6, ip_resolved_at, valid, updated_at
         FROM endpoints 
         WHERE endpoint = $1 AND endpoint_type = $2
     `
···
 
 	err := p.db.QueryRowContext(ctx, query, endpoint, endpointType).Scan(
 		&ep.ID, &ep.EndpointType, &ep.Endpoint, &ep.DiscoveredAt, &lastChecked,
-		&ep.Status, &ip, &ipv6, &ipResolvedAt, &ep.UpdatedAt,
+		&ep.Status, &ip, &ipv6, &ipResolvedAt, &ep.Valid, &ep.UpdatedAt,
 	)
 	if err != nil {
 		return nil, err
···
 	query := `
 	SELECT DISTINCT ON (COALESCE(server_did, id::text))
 		id, endpoint_type, endpoint, server_did, discovered_at, last_checked, status, 
-		ip, ipv6, ip_resolved_at, updated_at
+		ip, ipv6, ip_resolved_at, valid, updated_at
 	FROM endpoints
 	WHERE 1=1
 	`
···
 			query += fmt.Sprintf(" AND endpoint_type = $%d", argIdx)
 			args = append(args, filter.Type)
 			argIdx++
+		}
+
+		// NEW: Filter by valid flag
+		if filter.OnlyValid {
+			query += fmt.Sprintf(" AND valid = true", argIdx)
 		}
 		if filter.Status != "" {
 			statusInt := EndpointStatusUnknown
···
 				last_checked,
 				status,
 				ip,
-				ipv6
+				ipv6,
+				valid
 			FROM endpoints
 			WHERE endpoint_type = 'pds'
 			ORDER BY COALESCE(server_did, id::text), discovered_at ASC
 		)
 		SELECT 
-			e.id, e.endpoint, e.server_did, e.discovered_at, e.last_checked, e.status, e.ip, e.ipv6,
+			e.id, e.endpoint, e.server_did, e.discovered_at, e.last_checked, e.status, e.ip, e.ipv6, e.valid,
 			latest.user_count, latest.response_time, latest.version, latest.scanned_at,
 			i.city, i.country, i.country_code, i.asn, i.asn_org,
 			i.is_datacenter, i.is_vpn, i.is_crawler, i.is_tor, i.is_proxy,
···
 		var scannedAt sql.NullTime
 
 		err := rows.Scan(
-			&item.ID, &item.Endpoint, &serverDID, &item.DiscoveredAt, &item.LastChecked, &item.Status, &ip, &ipv6,
+			&item.ID, &item.Endpoint, &serverDID, &item.DiscoveredAt, &item.LastChecked, &item.Status, &ip, &ipv6, &item.Valid,
 			&userCount, &responseTime, &version, &scannedAt,
 			&city, &country, &countryCode, &asn, &asnOrg,
 			&isDatacenter, &isVPN, &isCrawler, &isTor, &isProxy,
···
 
 func (p *PostgresDB) GetPDSDetail(ctx context.Context, endpoint string) (*PDSDetail, error) {
 	query := `
-		WITH target_endpoint AS MATERIALIZED (  -- MATERIALIZED fence for optimization
+		WITH target_endpoint AS MATERIALIZED (
 			SELECT 
 				e.id, 
 				e.endpoint, 
···
 				e.last_checked, 
 				e.status, 
 				e.ip,
-				e.ipv6
+				e.ipv6,
+				e.valid
 			FROM endpoints e
 			WHERE e.endpoint = $1 
 			AND e.endpoint_type = 'pds'
-			LIMIT 1  -- Early termination since we expect exactly 1 row
+			LIMIT 1
 		)
 		SELECT 
 			te.id, 
···
 			te.status, 
 			te.ip,
 			te.ipv6,
+			te.valid,
 			latest.user_count,
 			latest.response_time,
 			latest.version,
···
 			i.is_datacenter, i.is_vpn, i.is_crawler, i.is_tor, i.is_proxy,
 			i.latitude, i.longitude,
 			i.raw_data,
-			-- Inline aliases aggregation (avoid second CTE)
 			COALESCE(
 				ARRAY(
 					SELECT e2.endpoint 
···
 				), 
 				ARRAY[]::text[]
 			) as aliases,
-			-- Inline first_discovered_at (avoid aggregation)
 			CASE 
 				WHEN te.server_did IS NOT NULL THEN (
 					SELECT MIN(e3.discovered_at)
···
 	var firstDiscoveredAt sql.NullTime
 
 	err := p.db.QueryRowContext(ctx, query, endpoint).Scan(
-		&detail.ID, &detail.Endpoint, &serverDID, &detail.DiscoveredAt, &detail.LastChecked, &detail.Status, &ip, &ipv6,
+		&detail.ID, &detail.Endpoint, &serverDID, &detail.DiscoveredAt, &detail.LastChecked, &detail.Status, &ip, &ipv6, &detail.Valid,
 		&userCount, &responseTime, &version, &serverInfoJSON, &scannedAt,
 		&city, &country, &countryCode, &asn, &asnOrg,
 		&isDatacenter, &isVPN, &isCrawler, &isTor, &isProxy,
···
 	// Set aliases and is_primary
 	detail.Aliases = aliases
 	if serverDID.Valid && serverDID.String != "" && firstDiscoveredAt.Valid {
-		// Has server_did - check if this is the first discovered
 		detail.IsPrimary = detail.DiscoveredAt.Equal(firstDiscoveredAt.Time) ||
 			detail.DiscoveredAt.Before(firstDiscoveredAt.Time)
 	} else {
-		// No server_did means unique server
 		detail.IsPrimary = true
 	}
 

internal/storage/types.go

···
 	LastChecked  time.Time
 	Status       int
 	IP           string
-	IPv6         string // NEW
+	IPv6         string
 	IPResolvedAt time.Time
+	Valid        bool
 	UpdatedAt    time.Time
 }
 
···
 
 // EndpointFilter for querying endpoints
 type EndpointFilter struct {
-	Type            string // "pds", "labeler", etc.
+	Type            string
 	Status          string
 	MinUserCount    int64
 	OnlyStale       bool
+	OnlyValid       bool
 	RecheckInterval time.Duration
-	Random          bool // NEW: Return results in random order
+	Random          bool
 	Limit           int
 	Offset          int
 }
···
 	LastChecked  time.Time
 	Status       int
 	IP           string
-	IPv6         string // NEW
+	IPv6         string
+	Valid        bool // NEW
 
 	// From latest endpoint_scans (via JOIN)
 	LatestScan *struct {