update

internal/api/handlers.go

···
 		return
 	}
 
-	respondJSON(w, pds)
+	// Optionally include DIDs list or just count
+	includeDIDs := r.URL.Query().Get("include_dids") == "true"
+
+	response := map[string]interface{}{
+		"endpoint":      pds.Endpoint,
+		"discovered_at": pds.DiscoveredAt,
+		"last_checked":  pds.LastChecked,
+		"status":        pds.Status,
+		"user_count":    pds.UserCount,
+	}
+
+	if includeDIDs {
+		response["dids"] = pds.DIDs
+	}
+
+	respondJSON(w, response)
 }
 
 func (s *Server) handleGetPDSStats(w http.ResponseWriter, r *http.Request) {
···
 	vars := mux.Vars(r)
 	did := vars["did"]
 
-	// Get all bundles and search for DID
-	bundles, err := s.db.GetAllBundles(ctx)
+	// Use SQL JSON query to find bundles
+	bundles, err := s.db.GetBundlesForDID(ctx, did)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
-	// Find bundles containing this DID
-	var relevantBundles []*storage.PLCBundle
-	for _, bundle := range bundles {
-		for _, bundleDID := range bundle.DIDs {
-			if bundleDID == did {
-				relevantBundles = append(relevantBundles, bundle)
-				break
-			}
-		}
-	}
-
-	if len(relevantBundles) == 0 {
+	if len(bundles) == 0 {
 		http.Error(w, "DID not found in bundles", http.StatusNotFound)
 		return
 	}
 
-	// Load the last bundle and find latest operation
-	lastBundle := relevantBundles[len(relevantBundles)-1]
+	// Load the last bundle (most recent)
+	lastBundle := bundles[len(bundles)-1]
 
 	operations, err := s.loadBundleOperations(lastBundle.FilePath)
 	if err != nil {
···
 	vars := mux.Vars(r)
 	did := vars["did"]
 
-	// Get all bundles
-	bundles, err := s.db.GetAllBundles(ctx)
+	// Use SQL JSON query
+	bundles, err := s.db.GetBundlesForDID(ctx, did)
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
 
-	// Find bundles containing this DID
-	var relevantBundles []*storage.PLCBundle
-	for _, bundle := range bundles {
-		for _, bundleDID := range bundle.DIDs {
-			if bundleDID == did {
-				relevantBundles = append(relevantBundles, bundle)
-				break
-			}
-		}
-	}
-
-	if len(relevantBundles) == 0 {
+	if len(bundles) == 0 {
 		http.Error(w, "DID not found in bundles", http.StatusNotFound)
 		return
 	}
···
 	var allOperations []plc.DIDHistoryEntry
 	var currentOp *plc.PLCOperation
 
-	// Load relevant bundles
-	for _, bundle := range relevantBundles {
+	// Load all relevant bundles
+	for _, bundle := range bundles {
 		operations, err := s.loadBundleOperations(bundle.FilePath)
 		if err != nil {
 			log.Printf("Warning: failed to load bundle: %v", err)
···
 		return
 	}
 
-	// Get all bundles and find the one we want
-	bundles, err := s.db.GetAllBundles(ctx)
+	bundle, err := s.db.GetBundleByID(ctx, bundleID)
 	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
+		http.Error(w, "bundle not found", http.StatusNotFound)
 		return
 	}
 
-	for _, bundle := range bundles {
-		if bundle.ID == bundleID {
-			respondJSON(w, map[string]interface{}{
-				"bundle_id": bundleID,
-				"did_count": len(bundle.DIDs),
-				"dids":      bundle.DIDs,
-			})
-			return
-		}
-	}
-
-	http.Error(w, "bundle not found", http.StatusNotFound)
+	respondJSON(w, map[string]interface{}{
+		"bundle_id": bundleID,
+		"did_count": len(bundle.DIDs),
+		"dids":      bundle.DIDs,
+	})
 }
 
 // Helper to load bundle operations

internal/pds/client.go

···
 	}
 }
 
+// ListReposResponse represents the response from com.atproto.sync.listRepos
+type ListReposResponse struct {
+	Repos  []Repo  `json:"repos"`
+	Cursor *string `json:"cursor,omitempty"`
+}
+
+// Repo represents a repository in the list
+type Repo struct {
+	DID  string `json:"did"`
+	Head string `json:"head,omitempty"`
+	Rev  string `json:"rev,omitempty"`
+}
+
+// ListRepos fetches all repositories from a PDS with pagination
+func (c *Client) ListRepos(ctx context.Context, endpoint string) ([]string, error) {
+	var allDIDs []string
+	var cursor *string
+
+	for {
+		// Build URL
+		url := fmt.Sprintf("%s/xrpc/com.atproto.sync.listRepos?limit=1000", endpoint)
+		if cursor != nil {
+			url += fmt.Sprintf("&cursor=%s", *cursor)
+		}
+
+		req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		resp, err := c.httpClient.Do(req)
+		if err != nil {
+			return nil, err
+		}
+
+		if resp.StatusCode != http.StatusOK {
+			resp.Body.Close()
+			return nil, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+		}
+
+		var result ListReposResponse
+		if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+			resp.Body.Close()
+			return nil, err
+		}
+		resp.Body.Close()
+
+		// Collect DIDs
+		for _, repo := range result.Repos {
+			allDIDs = append(allDIDs, repo.DID)
+		}
+
+		// Check if there are more pages
+		if result.Cursor == nil || *result.Cursor == "" {
+			break
+		}
+		cursor = result.Cursor
+	}
+
+	return allDIDs, nil
+}
+
 // DescribeServer fetches com.atproto.server.describeServer
 func (c *Client) DescribeServer(ctx context.Context, endpoint string) (*ServerDescription, error) {
 	url := fmt.Sprintf("%s/xrpc/com.atproto.server.describeServer", endpoint)

internal/pds/scanner.go

···
 	// Process results
 	successCount := 0
 	failureCount := 0
+	totalUsers := int64(0)
 
 	for status := range results {
 		if err := s.db.UpdatePDSStatus(ctx, status.Endpoint, &storage.PDSUpdate{
···
 			ResponseTime: status.ResponseTime.Milliseconds(),
 			ErrorMessage: status.ErrorMessage,
 			ServerInfo:   status.Description,
+			DIDs:         status.DIDs, // NEW: Store DIDs
 		}); err != nil {
 			log.Printf("Error updating PDS %s: %v", status.Endpoint, err)
 		}
 
 		if status.Available {
 			successCount++
+			totalUsers += int64(len(status.DIDs))
 		} else {
 			failureCount++
 		}
 	}
 
-	log.Printf("PDS scan completed: %d available, %d unavailable in %v",
-		successCount, failureCount, time.Since(startTime))
+	log.Printf("PDS scan completed: %d available, %d unavailable, %d total users in %v",
+		successCount, failureCount, totalUsers, time.Since(startTime))
 
 	return nil
 }
···
 	// Describe server
 	desc, err := s.client.DescribeServer(ctx, endpoint)
 	if err != nil {
-		log.Printf("Error describing server %s: %v", endpoint, err)
-		// Still mark as available if health check passed
+		log.Printf("Warning: failed to describe server %s: %v", endpoint, err)
 	} else {
 		status.Description = desc
+	}
+
+	// List repos (DIDs) - NEW
+	dids, err := s.client.ListRepos(ctx, endpoint)
+	if err != nil {
+		log.Printf("Warning: failed to list repos for %s: %v", endpoint, err)
+		status.DIDs = []string{} // Empty if failed
+	} else {
+		status.DIDs = dids
+		log.Printf("  → Found %d users on %s", len(dids), endpoint)
 	}
 
 	return status

internal/pds/types.go

···
 	LastChecked  time.Time
 	ErrorMessage string
 	Description  *ServerDescription
+	DIDs         []string
 }

internal/plc/scanner.go

···
 
 func (s *Scanner) processBatch(ctx context.Context, operations []PLCOperation) (int64, error) {
 	newPDSCount := int64(0)
-	seenInBatch := make(map[string]bool)
+	seenInBatch := make(map[string]*PLCOperation) // Store operation, not just bool
 
 	for _, op := range operations {
 		if op.IsNullified() {
···
 			continue
 		}
 
-		if seenInBatch[pdsEndpoint] {
-			continue
+		// Track first occurrence in this batch
+		if _, seen := seenInBatch[pdsEndpoint]; !seen {
+			seenInBatch[pdsEndpoint] = &op
 		}
-		seenInBatch[pdsEndpoint] = true
+	}
 
+	// Now process unique PDS endpoints
+	for pdsEndpoint, firstOp := range seenInBatch {
 		exists, err := s.db.PDSExists(ctx, pdsEndpoint)
 		if err != nil {
 			continue
···
 			continue
 		}
 
+		// New PDS! Use the operation timestamp as discovered_at
 		if err := s.db.UpsertPDS(ctx, &storage.PDS{
 			Endpoint:     pdsEndpoint,
-			DiscoveredAt: time.Now(),
+			DiscoveredAt: firstOp.CreatedAt, // Use operation timestamp!
 			LastChecked:  time.Time{},
 			Status:       "unknown",
 		}); err != nil {
···
 			continue
 		}
 
-		log.Printf("✓ Discovered new PDS: %s", pdsEndpoint)
+		log.Printf("✓ Discovered new PDS: %s (first seen: %s)",
+			pdsEndpoint, firstOp.CreatedAt.Format("2006-01-02 15:04"))
 		newPDSCount++
 	}
 

internal/storage/db.go

···
 	// Bundle operations
 	CreateBundle(ctx context.Context, bundle *PLCBundle) error
 	GetBundle(ctx context.Context, afterTime time.Time) (*PLCBundle, error)
+	GetBundleByID(ctx context.Context, bundleID int64) (*PLCBundle, error)
 	GetBundles(ctx context.Context, limit int) ([]*PLCBundle, error)
+	GetBundlesForDID(ctx context.Context, did string) ([]*PLCBundle, error) // JSON query
 	GetBundleStats(ctx context.Context) (int64, int64, error)
-	GetAllBundles(ctx context.Context) ([]*PLCBundle, error) // For DID search
 
 	// Metrics
 	StorePLCMetrics(ctx context.Context, metrics *PLCMetrics) error

internal/storage/sqlite.go

···
         response_time_ms INTEGER,
         error_message TEXT,
         server_info TEXT,
+        dids TEXT DEFAULT '[]',
         user_count INTEGER DEFAULT 0,
         updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
     );
 
     CREATE INDEX IF NOT EXISTS idx_pds_status ON pds_servers(status);
     CREATE INDEX IF NOT EXISTS idx_pds_last_checked ON pds_servers(last_checked);
+    CREATE INDEX IF NOT EXISTS idx_pds_user_count ON pds_servers(user_count);
 
     CREATE TABLE IF NOT EXISTS plc_metrics (
         id INTEGER PRIMARY KEY AUTOINCREMENT,
···
         created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
     );
 
-    CREATE INDEX IF NOT EXISTS idx_plc_bundles_time ON plc_bundles(start_time, end_time);
-    CREATE INDEX IF NOT EXISTS idx_plc_bundles_created ON plc_bundles(created_at);
+    -- CREATE INDEX IF NOT EXISTS idx_plc_bundles_time ON plc_bundles(start_time, end_time);
+    -- CREATE INDEX IF NOT EXISTS idx_plc_bundles_created ON plc_bundles(created_at);
     `
 
 	_, err := s.db.Exec(schema)
···
 	return s.scanBundles(rows)
 }
 
+// GetBundlesForDID finds bundles containing a specific DID using JSON functions
+func (s *SQLiteDB) GetBundlesForDID(ctx context.Context, did string) ([]*PLCBundle, error) {
+	query := `
+        SELECT id, start_time, end_time, operation_count, dids, file_path, file_size, compressed, created_at
+        FROM plc_bundles
+        WHERE EXISTS (
+            SELECT 1 FROM json_each(dids) 
+            WHERE json_each.value = ?
+        )
+        ORDER BY start_time ASC
+    `
+
+	rows, err := s.db.QueryContext(ctx, query, did)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	return s.scanBundles(rows)
+}
+
+// GetBundleByID retrieves a single bundle by ID
+func (s *SQLiteDB) GetBundleByID(ctx context.Context, bundleID int64) (*PLCBundle, error) {
+	query := `
+        SELECT id, start_time, end_time, operation_count, dids, file_path, file_size, compressed, created_at
+        FROM plc_bundles
+        WHERE id = ?
+    `
+
+	var bundle PLCBundle
+	var didsJSON string
+
+	err := s.db.QueryRowContext(ctx, query, bundleID).Scan(
+		&bundle.ID, &bundle.StartTime, &bundle.EndTime, &bundle.OperationCount,
+		&didsJSON, &bundle.FilePath, &bundle.FileSize, &bundle.Compressed, &bundle.CreatedAt,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := json.Unmarshal([]byte(didsJSON), &bundle.DIDs); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal DIDs: %w", err)
+	}
+
+	return &bundle, nil
+}
+
 // Helper to scan bundle rows
 func (s *SQLiteDB) scanBundles(rows *sql.Rows) ([]*PLCBundle, error) {
 	var bundles []*PLCBundle
···
 	return err
 }
 
+// UpdatePDSStatus updates the status, metrics, and DIDs of a PDS server
+func (s *SQLiteDB) UpdatePDSStatus(ctx context.Context, endpoint string, update *PDSUpdate) error {
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	// Marshal server info
+	var serverInfoJSON []byte
+	if update.ServerInfo != nil {
+		serverInfoJSON, _ = json.Marshal(update.ServerInfo)
+	}
+
+	// Marshal DIDs
+	var didsJSON []byte
+	if update.DIDs != nil {
+		didsJSON, _ = json.Marshal(update.DIDs)
+	} else {
+		didsJSON = []byte("[]")
+	}
+
+	// Calculate user count
+	userCount := len(update.DIDs)
+
+	query := `
+        UPDATE pds_servers 
+        SET status = ?, last_checked = ?, response_time_ms = ?, error_message = ?, 
+            server_info = ?, dids = ?, user_count = ?, updated_at = ?
+        WHERE endpoint = ?
+    `
+	_, err = tx.ExecContext(ctx, query,
+		update.Status, update.LastChecked, update.ResponseTime,
+		update.ErrorMessage, string(serverInfoJSON), string(didsJSON), userCount, time.Now(), endpoint)
+	if err != nil {
+		return err
+	}
+
+	// Insert scan history
+	scanQuery := `
+        INSERT INTO pds_scans (endpoint, status, response_time_ms, error_message, server_info)
+        VALUES (?, ?, ?, ?, ?)
+    `
+	_, err = tx.ExecContext(ctx, scanQuery, endpoint, update.Status, update.ResponseTime,
+		update.ErrorMessage, string(serverInfoJSON))
+	if err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
 // GetPDS retrieves a single PDS server by endpoint
 func (s *SQLiteDB) GetPDS(ctx context.Context, endpoint string) (*PDS, error) {
 	query := `
         SELECT endpoint, discovered_at, last_checked, status, response_time_ms, 
-               error_message, server_info, user_count 
+               error_message, server_info, dids, user_count 
         FROM pds_servers 
         WHERE endpoint = ?
     `
···
 	var responseTime sql.NullInt64
 	var errorMsg sql.NullString
 	var serverInfo sql.NullString
+	var didsJSON string
 
 	err := s.db.QueryRowContext(ctx, query, endpoint).Scan(
 		&pds.Endpoint, &pds.DiscoveredAt, &lastChecked, &pds.Status,
-		&responseTime, &errorMsg, &serverInfo, &pds.UserCount,
+		&responseTime, &errorMsg, &serverInfo, &didsJSON, &pds.UserCount,
 	)
 	if err != nil {
 		return nil, err
···
 	}
 	if serverInfo.Valid && serverInfo.String != "" {
 		json.Unmarshal([]byte(serverInfo.String), &pds.ServerInfo)
+	}
+
+	// Unmarshal DIDs
+	if didsJSON != "" {
+		json.Unmarshal([]byte(didsJSON), &pds.DIDs)
 	}
 
 	return &pds, nil
···
 	}
 
 	return servers, rows.Err()
-}
-
-// UpdatePDSStatus updates the status and metrics of a PDS server
-func (s *SQLiteDB) UpdatePDSStatus(ctx context.Context, endpoint string, update *PDSUpdate) error {
-	tx, err := s.db.BeginTx(ctx, nil)
-	if err != nil {
-		return err
-	}
-	defer tx.Rollback()
-
-	// Update main record
-	var serverInfoJSON []byte
-	if update.ServerInfo != nil {
-		serverInfoJSON, _ = json.Marshal(update.ServerInfo)
-	}
-
-	query := `
-        UPDATE pds_servers 
-        SET status = ?, last_checked = ?, response_time_ms = ?, error_message = ?, server_info = ?, updated_at = ?
-        WHERE endpoint = ?
-    `
-	_, err = tx.ExecContext(ctx, query, update.Status, update.LastChecked, update.ResponseTime,
-		update.ErrorMessage, string(serverInfoJSON), time.Now(), endpoint)
-	if err != nil {
-		return err
-	}
-
-	// Insert scan history
-	scanQuery := `
-        INSERT INTO pds_scans (endpoint, status, response_time_ms, error_message, server_info)
-        VALUES (?, ?, ?, ?, ?)
-    `
-	_, err = tx.ExecContext(ctx, scanQuery, endpoint, update.Status, update.ResponseTime,
-		update.ErrorMessage, string(serverInfoJSON))
-	if err != nil {
-		return err
-	}
-
-	return tx.Commit()
 }
 
 // GetScanCursor retrieves the last scan cursor for a source

internal/storage/types.go

···
 	ResponseTime int64
 	ErrorMessage string
 	ServerInfo   interface{}
-	UserCount    int64
+	DIDs         []string // NEW: List of DIDs hosted on this PDS
+	UserCount    int64    // Calculated from len(DIDs)
 }
 
 // PDSUpdate contains fields to update for a PDS
···
 	ResponseTime int64
 	ErrorMessage string
 	ServerInfo   interface{}
+	DIDs         []string
 }
 
 // PDSFilter for querying PDS servers

utils/plc-cache-size.sh

···
+#!/bin/sh
+total_uncompressed_size=0
+for file in plc_cache/*.zst; do
+  uncompressed_size=$(zstd -l "$file" | awk '/Decompressed Size/ {print $3}')
+  total_uncompressed_size=$((total_uncompressed_size + uncompressed_size))
+done
+echo "Total uncompressed size: $total_uncompressed_size Bytes"